diff --git a/.github/aicodingflow-tests/test_context_helpers.py b/.github/aicodingflow-tests/test_context_helpers.py new file mode 100644 index 0000000..375a543 --- /dev/null +++ b/.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() diff --git a/.github/aicodingflow-tests/test_github_api.py b/.github/aicodingflow-tests/test_github_api.py new file mode 100644 index 0000000..c604c6a --- /dev/null +++ b/.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() diff --git a/.github/aicodingflow-tests/test_handle_plan_approved.py b/.github/aicodingflow-tests/test_handle_plan_approved.py index 4be0d7a..b1c73d8 100644 --- a/.github/aicodingflow-tests/test_handle_plan_approved.py +++ b/.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( diff --git a/.github/aicodingflow-tests/test_issue_refs.py b/.github/aicodingflow-tests/test_issue_refs.py new file mode 100644 index 0000000..69def5d --- /dev/null +++ b/.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() diff --git a/.github/aicodingflow-tests/test_ledger_contracts.py b/.github/aicodingflow-tests/test_ledger_contracts.py new file mode 100644 index 0000000..c4465d7 --- /dev/null +++ b/.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() diff --git a/.github/aicodingflow-tests/test_pr_metadata_contracts.py b/.github/aicodingflow-tests/test_pr_metadata_contracts.py new file mode 100644 index 0000000..fbbd144 --- /dev/null +++ b/.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() diff --git a/.github/aicodingflow-tests/test_prepare_local_review_inputs.py b/.github/aicodingflow-tests/test_prepare_local_review_inputs.py index 67e75d0..36257dd 100644 --- a/.github/aicodingflow-tests/test_prepare_local_review_inputs.py +++ b/.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")) diff --git a/.github/scripts/apply_issue_triage_result.py b/.github/scripts/apply_issue_triage_result.py index a232501..35d5369 100644 --- a/.github/scripts/apply_issue_triage_result.py +++ b/.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 = "" 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")) diff --git a/.github/scripts/apply_pr_comment_result.py b/.github/scripts/apply_pr_comment_result.py index 12ffab1..377401a 100644 --- a/.github/scripts/apply_pr_comment_result.py +++ b/.github/scripts/apply_pr_comment_result.py @@ -9,6 +9,9 @@ from pathlib import Path from typing import Any +from artifact_contracts import write_github_output +from github_api import flatten_gh_pages as flatten_pages + PAGE_SIZE = 100 PAGE_INFO = "pageInfo { hasNextPage endCursor }" @@ -46,14 +49,6 @@ def load_json(path: Path, *, required: bool = True) -> dict[str, Any]: return value -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 open_pr_for_branch(repo: str, branch_name: str) -> dict[str, Any] | None: owner = repo.split("/", 1)[0] pages = run_gh_json( @@ -237,14 +232,6 @@ def apply_result(repo: str, context: dict[str, Any], metadata: dict[str, Any], r return {"pr_url": pr_url, "resolve_warnings": "\n".join(warnings)} -def write_github_output(path: str | None, values: dict[str, str]) -> None: - if not path: - return - with Path(path).open("a", encoding="utf-8") as handle: - for key, value in values.items(): - handle.write(f"{key}={value}\n") - - def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--repo", required=True) diff --git a/.github/scripts/artifact_contracts.py b/.github/scripts/artifact_contracts.py new file mode 100644 index 0000000..eb6f24e --- /dev/null +++ b/.github/scripts/artifact_contracts.py @@ -0,0 +1,25 @@ +"""Shared artifact helpers for GitHub workflow scripts.""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + + +def load_json(path: str | Path | None, *, default: Any = None) -> Any: + if not path: + return default + return json.loads(Path(path).read_text(encoding="utf-8")) + + +def write_json(path: str | Path, payload: Any, *, ensure_ascii: bool = False) -> None: + Path(path).write_text(json.dumps(payload, ensure_ascii=ensure_ascii, indent=2) + "\n", encoding="utf-8") + + +def write_github_output(path: str | Path | None, values: dict[str, str]) -> None: + if not path: + return + with Path(path).open("a", encoding="utf-8") as handle: + for key, value in values.items(): + handle.write(f"{key}={value}\n") diff --git a/.github/scripts/commit_implementation_branch.py b/.github/scripts/commit_implementation_branch.py index 43c8c44..50a87e3 100644 --- a/.github/scripts/commit_implementation_branch.py +++ b/.github/scripts/commit_implementation_branch.py @@ -10,6 +10,7 @@ from pathlib import Path from typing import Any +from artifact_contracts import write_github_output from implementation_file_filters import TEMP_WORKFLOW_PATHS, is_generated_path, is_github_workflow_path @@ -235,14 +236,6 @@ def commit_and_push(context_path: Path, metadata_path: Path, author_name: str, a return {"changed": "true", "branch": branch, "sha": sha} -def write_github_output(path: str | None, values: dict[str, str]) -> None: - if not path: - return - with Path(path).open("a", encoding="utf-8") as handle: - for key, value in values.items(): - handle.write(f"{key}={value}\n") - - def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--context", default="issue_context.json") diff --git a/.github/scripts/context_snapshot.py b/.github/scripts/context_snapshot.py new file mode 100644 index 0000000..c51fe98 --- /dev/null +++ b/.github/scripts/context_snapshot.py @@ -0,0 +1,50 @@ +"""Shared stable context snapshot helpers.""" + +from __future__ import annotations + +from typing import Any + +from github_event import actor_login + + +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 remove_triggering_comment( + comments: list[dict[str, Any]], + trigger_comment: dict[str, Any] | None, +) -> list[dict[str, Any]]: + if not trigger_comment: + return comments + trigger_id = trigger_comment.get("id") + trigger_url = trigger_comment.get("url") + filtered: list[dict[str, Any]] = [] + for comment in comments: + if trigger_id is not None and comment.get("id") == trigger_id: + continue + if trigger_url and comment.get("html_url") == trigger_url: + continue + filtered.append(comment) + return filtered + + +def format_issue_comments(comments: list[dict[str, Any]]) -> str: + lines: list[str] = [] + for comment in comments: + lines.extend( + [ + f"Author: {actor_login(comment)}", + f"Created: {comment.get('created_at') or ''}", + "", + comment.get("body") or "", + "", + "---", + "", + ] + ) + return "\n".join(lines) diff --git a/.github/scripts/finalize_implementation_pr.py b/.github/scripts/finalize_implementation_pr.py index 3483a73..962eaab 100644 --- a/.github/scripts/finalize_implementation_pr.py +++ b/.github/scripts/finalize_implementation_pr.py @@ -9,18 +9,9 @@ from pathlib import Path from typing import Any - -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") +from artifact_contracts import write_github_output +from github_api import flatten_gh_pages as flatten_pages +from github_api import run_gh_json def open_pr_for_branch(repo: str, branch_name: str) -> dict[str, Any] | None: @@ -97,14 +88,6 @@ def finalize(repo: str, context: dict[str, Any], metadata: dict[str, Any]) -> st return create_pr(repo, str(context["default_branch"]), branch_name, title, body) -def write_github_output(path: str | None, values: dict[str, str]) -> None: - if not path: - return - with Path(path).open("a", encoding="utf-8") as handle: - for key, value in values.items(): - handle.write(f"{key}={value}\n") - - def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--repo", required=True) diff --git a/.github/scripts/finalize_spec_pr.py b/.github/scripts/finalize_spec_pr.py index ba34918..dcfe68f 100644 --- a/.github/scripts/finalize_spec_pr.py +++ b/.github/scripts/finalize_spec_pr.py @@ -9,6 +9,10 @@ from pathlib import Path from typing import Any +from artifact_contracts import write_github_output +from github_api import flatten_gh_pages as flatten_pages +from github_api import run_gh_json + def run(args: list[str], *, capture: bool = False, check: bool = True) -> str: result = subprocess.run( @@ -25,18 +29,6 @@ def load_json(path: Path) -> dict[str, Any]: return json.loads(path.read_text(encoding="utf-8")) -def run_gh_json(args: list[str]) -> Any: - return json.loads(run(["gh", *args], capture=True)) - - -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 open_pr_for_branch(repo: str, branch_name: str) -> dict[str, Any] | None: owner = repo.split("/", 1)[0] pages = run_gh_json( @@ -117,14 +109,6 @@ def create_or_update_pr(repo: str, context: dict[str, Any], metadata: dict[str, return create_pr(repo, str(context["default_branch"]), branch, title, body) -def write_github_output(path: str | None, values: dict[str, str]) -> None: - if not path: - return - with Path(path).open("a", encoding="utf-8") as handle: - for key, value in values.items(): - handle.write(f"{key}={value}\n") - - def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--repo", required=True) diff --git a/.github/scripts/github_api.py b/.github/scripts/github_api.py new file mode 100644 index 0000000..045b2d4 --- /dev/null +++ b/.github/scripts/github_api.py @@ -0,0 +1,32 @@ +"""Small GitHub CLI helpers shared by workflow scripts.""" + +from __future__ import annotations + +import json +import subprocess +from collections.abc import Callable +from typing import Any + +from context_snapshot import flatten_pages + + +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 run_gh_text(args: list[str]) -> str: + result = subprocess.run(["gh", *args], check=True, stdout=subprocess.PIPE, text=True) + return result.stdout + + +def fetch_default_branch(repo: str, *, run_json: Callable[[list[str]], Any] | None = None) -> str: + data = (run_json or run_gh_json)(["repo", "view", repo, "--json", "defaultBranchRef"]) + branch = (data.get("defaultBranchRef") or {}).get("name") + if not branch: + raise SystemExit("could not determine default branch") + return branch + + +def flatten_gh_pages(value: Any) -> list[dict[str, Any]]: + return flatten_pages(value) diff --git a/.github/scripts/github_event.py b/.github/scripts/github_event.py new file mode 100644 index 0000000..7d6fb4d --- /dev/null +++ b/.github/scripts/github_event.py @@ -0,0 +1,77 @@ +"""Shared GitHub event field helpers for workflow context scripts.""" + +from __future__ import annotations + +from typing import Any + + +def actor_login(item: dict[str, Any]) -> str: + user = item.get("user") or item.get("author") or {} + if not isinstance(user, dict): + return "" + return user.get("login") or "" + + +def label_names(issue: dict[str, Any]) -> list[str]: + return [label.get("name", "") for label in issue.get("labels", []) if isinstance(label, dict) and label.get("name")] + + +def assignee_logins(issue: dict[str, Any]) -> list[str]: + return [ + assignee.get("login", "") + for assignee in issue.get("assignees", []) + if isinstance(assignee, dict) and assignee.get("login") + ] + + +def event_action(event: dict[str, Any]) -> str: + return event.get("action") or "" + + +def event_issue(event: dict[str, Any]) -> dict[str, Any]: + issue = event.get("issue") + return issue if isinstance(issue, dict) else {} + + +def event_comment(event: dict[str, Any]) -> dict[str, Any]: + comment = event.get("comment") + return comment if isinstance(comment, dict) else {} + + +def event_comment_body(event: dict[str, Any]) -> str: + return event_comment(event).get("body") or "" + + +def event_label_name(event: dict[str, Any]) -> str: + label = event.get("label") or {} + if not isinstance(label, dict): + return "" + return label.get("name") or "" + + +def event_assignee_login(event: dict[str, Any]) -> str: + assignee = event.get("assignee") or {} + if not isinstance(assignee, dict): + return "" + return assignee.get("login") or "" + + +def is_pull_request_issue_event(event: dict[str, Any]) -> bool: + issue = event_issue(event) + return "pull_request" in issue and issue.get("pull_request") is not None + + +def triggering_comment_snapshot(event: dict[str, Any], *, include_author_association: bool = False) -> dict[str, Any] | None: + comment = event_comment(event) + if not comment: + return None + snapshot = { + "id": comment.get("id"), + "author": actor_login(comment), + "body": comment.get("body") or "", + "created_at": comment.get("created_at") or "", + "url": comment.get("html_url") or "", + } + if include_author_association: + snapshot["author_association"] = comment.get("author_association") or "" + return snapshot diff --git a/.github/scripts/handle_plan_approved.py b/.github/scripts/handle_plan_approved.py index 58c7fee..192edcb 100644 --- a/.github/scripts/handle_plan_approved.py +++ b/.github/scripts/handle_plan_approved.py @@ -4,28 +4,20 @@ from __future__ import annotations import argparse -import json import os -import re import subprocess -from pathlib import Path from typing import Any +from artifact_contracts import load_json, write_github_output +from github_api import fetch_default_branch, run_gh_json +from github_event import assignee_logins, label_names +from issue_refs import issue_number_from_branch, issue_number_from_strict_text + APPROVED_LABEL = "plan-approved" READY_TO_SPEC_LABEL = "ready-to-spec" READY_TO_IMPLEMENT_LABEL = "ready-to-implement" IMPLEMENTATION_WORKFLOW = "create-implementation-from-issue.yml" -SPEC_BRANCH_RE = re.compile(r"^spec/issue-(\d+)$") -LINKED_ISSUE_PATTERNS = [ - re.compile(r"(? Any: - result = subprocess.run(["gh", *args], check=True, stdout=subprocess.PIPE, text=True) - return json.loads(result.stdout) def run_gh(args: list[str]) -> None: @@ -33,25 +25,7 @@ def run_gh(args: list[str]) -> None: def load_event(path: str | None) -> dict[str, Any]: - if not path: - return {} - return json.loads(Path(path).read_text(encoding="utf-8")) - - -def write_github_output(path: str | None, values: dict[str, str]) -> None: - if not path: - return - with Path(path).open("a", encoding="utf-8") as handle: - for key, value in values.items(): - handle.write(f"{key}={value}\n") - - -def label_names(item: dict[str, Any]) -> list[str]: - return [label.get("name", "") for label in item.get("labels", []) if label.get("name")] - - -def assignee_logins(issue: dict[str, Any]) -> list[str]: - return [assignee.get("login", "") for assignee in issue.get("assignees", []) if assignee.get("login")] + return load_json(path, default={}) def head_ref(pr: dict[str, Any]) -> str: @@ -66,18 +40,11 @@ def head_repo_full_name(pr: dict[str, Any]) -> str: def issue_number_from_explicit_text(text: str) -> int | None: - for pattern in LINKED_ISSUE_PATTERNS: - match = pattern.search(text or "") - if match: - return int(match.group(1)) - return None + return issue_number_from_strict_text(text) def issue_number_from_spec_branch(branch: str) -> int | None: - match = SPEC_BRANCH_RE.match(branch or "") - if not match: - return None - return int(match.group(1)) + return issue_number_from_branch(branch, prefix="spec/issue-", include_suffix=False) def resolve_linked_issue_number(pr: dict[str, Any]) -> int | None: @@ -120,14 +87,6 @@ def fetch_issue(repo: str, issue_number: int) -> dict[str, Any]: ) -def fetch_default_branch(repo: str) -> str: - repository = run_gh_json(["repo", "view", repo, "--json", "defaultBranchRef"]) - default_branch = (repository.get("defaultBranchRef") or {}).get("name") - if not default_branch: - raise SystemExit("could not determine default branch") - return default_branch - - def remove_ready_to_spec_label(repo: str, issue_number: int, dry_run: bool = False) -> bool: if dry_run: return True @@ -186,23 +145,49 @@ def repository_default_branch(event: dict[str, Any]) -> str: return repository.get("default_branch") or "" -def handle_plan_approved(args: argparse.Namespace) -> dict[str, str]: - event = load_event(args.event_path) +def resolve_event_pr(args: argparse.Namespace, event: dict[str, Any]) -> dict[str, Any] | None: pr = event_pull_request(event) - if not pr: - pr_number = int(args.pr_number) if args.pr_number else event_pr_number(event) - if not pr_number: - return build_outputs(skip_reason="pull request is not available") - pr = fetch_pr(args.repo, pr_number) + if pr: + return pr + pr_number = int(args.pr_number) if args.pr_number else event_pr_number(event) + return fetch_pr(args.repo, pr_number) if pr_number else None + +def validate_spec_pr(pr: dict[str, Any]) -> tuple[int | None, str]: issue_number = resolve_linked_issue_number(pr) if issue_number is None: - return build_outputs(skip_reason="linked issue not found") + return None, "linked issue not found" if not is_spec_pr(pr, issue_number): - return build_outputs(issue_number=issue_number, skip_reason="pull request is not a spec PR") - + return issue_number, "pull request is not a spec PR" if APPROVED_LABEL not in label_names(pr): - return build_outputs(issue_number=issue_number, skip_reason="pull request is missing plan-approved") + return issue_number, "pull request is missing plan-approved" + return issue_number, "" + + +def implementation_skip_reason( + *, + has_ready_to_implement: bool, + agent_login: str, + has_agent_assignee: bool, +) -> str: + if not has_ready_to_implement: + return f"missing {READY_TO_IMPLEMENT_LABEL}" + if not agent_login: + return "missing agent login" + if not has_agent_assignee: + return "missing bot assignee" + return "" + + +def handle_plan_approved(args: argparse.Namespace) -> dict[str, str]: + event = load_event(args.event_path) + pr = resolve_event_pr(args, event) + if not pr: + return build_outputs(skip_reason="pull request is not available") + + issue_number, skip_reason = validate_spec_pr(pr) + if skip_reason: + return build_outputs(issue_number=issue_number, skip_reason=skip_reason) issue = fetch_issue(args.repo, issue_number) issue_labels = set(label_names(issue)) @@ -217,17 +202,14 @@ def handle_plan_approved(args: argparse.Namespace) -> dict[str, str]: has_agent_assignee = bool(agent_login and agent_login in issue_assignees) default_branch = repository_default_branch(event) or fetch_default_branch(args.repo) - skip_reason = "" - implementation_dispatched = False - if not has_ready_to_implement: - skip_reason = f"missing {READY_TO_IMPLEMENT_LABEL}" - elif not agent_login: - skip_reason = "missing agent login" - elif not has_agent_assignee: - skip_reason = "missing bot assignee" - else: + skip_reason = implementation_skip_reason( + has_ready_to_implement=has_ready_to_implement, + agent_login=agent_login, + has_agent_assignee=has_agent_assignee, + ) + implementation_dispatched = not skip_reason + if implementation_dispatched: dispatch_implementation(args.repo, default_branch, issue_number, agent_login, args.dry_run) - implementation_dispatched = True return build_outputs( issue_number=issue_number, diff --git a/.github/scripts/issue_refs.py b/.github/scripts/issue_refs.py new file mode 100644 index 0000000..b13f381 --- /dev/null +++ b/.github/scripts/issue_refs.py @@ -0,0 +1,93 @@ +"""Shared issue reference parsing helpers.""" + +from __future__ import annotations + +import re +from typing import Any + + +EXPLICIT_ISSUE_PATTERNS = [ + re.compile(r"(? int | None: + value = text or "" + for pattern in EXPLICIT_ISSUE_PATTERNS: + match = pattern.search(value) + if match: + return int(match.group(1)) + if include_branch_suffix: + match = BRANCH_SUFFIX_RE.search(value) + if match: + return int(match.group(1)) + return None + + +def issue_number_from_strict_text(text: str) -> int | None: + value = text or "" + for pattern in STRICT_ISSUE_PATTERNS: + match = pattern.search(value) + if match: + return int(match.group(1)) + return None + + +def issue_number_from_branch(branch: str, *, prefix: str = "spec/issue-", include_suffix: bool = True) -> int | None: + value = branch or "" + if prefix and value.startswith(prefix): + suffix = value.removeprefix(prefix) + return int(suffix) if suffix.isdigit() else None + if not include_suffix: + return None + match = BRANCH_SUFFIX_RE.search(value) + return int(match.group(1)) if match else None + + +def resolve_issue_number_from_pr(pr: dict[str, Any]) -> int | None: + for text in ( + pr.get("body") or "", + pr.get("title") or "", + (pr.get("head") or {}).get("ref") or pr.get("headRefName") or "", + ): + issue_number = issue_number_from_text(str(text)) + if issue_number is not None: + return issue_number + return None + + +def issue_numbers_from_closing_refs(pr: dict[str, Any]) -> list[int]: + numbers: list[int] = [] + + def add_number(value: Any) -> None: + try: + number = int(value) + except (TypeError, ValueError): + return + if number not in numbers: + numbers.append(number) + + for issue in pr.get("closingIssuesReferences") or []: + if isinstance(issue, dict): + add_number(issue.get("number")) + for text in (pr.get("title") or "", pr.get("body") or ""): + for match in ISSUE_REFERENCE_KEYWORD_RE.finditer(str(text)): + reference_text = match.group(0) + for number_match in ISSUE_NUMBER_RE.finditer(reference_text): + if PULL_REQUEST_REFERENCE_PREFIX_RE.search(reference_text[: number_match.start()]): + continue + add_number(number_match.group(1)) + return numbers diff --git a/.github/scripts/ledger_contracts.py b/.github/scripts/ledger_contracts.py new file mode 100644 index 0000000..d89f305 --- /dev/null +++ b/.github/scripts/ledger_contracts.py @@ -0,0 +1,58 @@ +"""Shared helpers for PR-keyed workflow ledgers.""" + +from __future__ import annotations + +import datetime as dt +from pathlib import Path +from typing import Any + +from artifact_contracts import load_json, write_json + + +UTC = dt.timezone.utc + + +def now_iso() -> str: + return dt.datetime.now(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z") + + +def load_ledger(path: Path, *, ledger_name: str) -> dict[str, Any]: + if not path.exists(): + return {"version": 1, "entries": []} + data = load_json(path) + if not isinstance(data, dict): + raise SystemExit(f"invalid {ledger_name} ledger: {path}") + data.setdefault("version", 1) + data.setdefault("entries", []) + if not isinstance(data["entries"], list): + raise SystemExit(f"invalid {ledger_name} ledger entries: {path}") + return data + + +def entries_by_pr(ledger: dict[str, Any]) -> dict[int, dict[str, Any]]: + entries: dict[int, dict[str, Any]] = {} + for entry in ledger.get("entries") or []: + try: + entries[int(entry.get("pr"))] = entry + except (TypeError, ValueError): + continue + return entries + + +def merge_commit_oid(pr: dict[str, Any]) -> str: + merge_commit = pr.get("mergeCommit") or {} + return merge_commit.get("oid") or "" + + +def sort_entries(entries: list[dict[str, Any]]) -> list[dict[str, Any]]: + return sorted(entries, key=lambda item: (item.get("merged_at") or "", int(item.get("pr") or 0))) + + +def set_sorted_entries(ledger: dict[str, Any], entries: list[dict[str, Any]]) -> dict[str, Any]: + ledger["version"] = 1 + ledger["entries"] = sort_entries(entries) + return ledger + + +def write_ledger(path: Path, ledger: dict[str, Any]) -> None: + write_json(path, ledger) diff --git a/.github/scripts/pr_metadata_contracts.py b/.github/scripts/pr_metadata_contracts.py new file mode 100644 index 0000000..b20c584 --- /dev/null +++ b/.github/scripts/pr_metadata_contracts.py @@ -0,0 +1,92 @@ +"""Shared validation helpers for agent-authored PR metadata.""" + +from __future__ import annotations + +import json +import re +from pathlib import Path +from typing import Any + +from implementation_file_filters import TEMP_WORKFLOW_PATHS, is_generated_path + + +CONVENTIONAL_TITLE_RE = re.compile(r"^(feat|fix|docs|style|refactor|perf|test|build|ci|chore)(\([a-z0-9._-]+\))?: .+") +BASE_METADATA_FIELDS = {"branch_name", "pr_title", "pr_summary"} +INTENDED_FILES_FIELD = "intended_files" + + +def load_json_object(path: Path, *, required: bool = True, display_name: str | None = None) -> dict[str, Any]: + name = display_name or str(path) + if not path.exists(): + if required: + raise SystemExit(f"{name} was not created") + return {} + try: + value = json.loads(path.read_text(encoding="utf-8")) + except json.JSONDecodeError as exc: + raise SystemExit(f"{name} is invalid JSON: {exc}") from exc + if not isinstance(value, dict): + raise SystemExit(f"{name} must contain a JSON object") + return value + + +def require_fields(metadata: dict[str, Any], required: set[str], *, display_name: str = "pr-metadata.json") -> None: + missing = sorted(required - set(metadata)) + if missing: + raise SystemExit(f"{display_name} is missing fields: {', '.join(missing)}") + + +def require_non_empty_strings( + metadata: dict[str, Any], + fields: set[str], + *, + display_name: str = "pr-metadata.json", +) -> None: + for field in fields: + if not isinstance(metadata.get(field), str) or not metadata[field].strip(): + raise SystemExit(f"{display_name} field {field} must be a non-empty string") + + +def require_conventional_title(title: str, *, display_name: str = "pr-metadata.json") -> None: + if not CONVENTIONAL_TITLE_RE.match(title): + raise SystemExit(f"{display_name} pr_title must use conventional commit style") + + +def require_markdown_summary(summary: str, *, display_name: str = "pr-metadata.json") -> None: + if "\n" not in summary: + raise SystemExit(f"{display_name} pr_summary must be a complete markdown body, not a one-line note") + + +def validate_base_metadata( + metadata: dict[str, Any], + *, + required_fields: set[str] | None = None, + string_fields: set[str] | None = None, + display_name: str = "pr-metadata.json", +) -> None: + required = required_fields or BASE_METADATA_FIELDS + strings = string_fields or BASE_METADATA_FIELDS + require_fields(metadata, required, display_name=display_name) + require_non_empty_strings(metadata, strings, display_name=display_name) + require_conventional_title(metadata["pr_title"], display_name=display_name) + require_markdown_summary(metadata["pr_summary"], display_name=display_name) + + +def validate_intended_path(path: object, index: int, *, display_name: str = "pr-metadata.json") -> str: + if not isinstance(path, str) or not path.strip(): + raise SystemExit(f"{display_name} intended_files[{index}] must be a non-empty string") + normalized = path.strip() + if Path(normalized).is_absolute() or ".." in Path(normalized).parts: + raise SystemExit(f"{display_name} intended_files[{index}] must be a repository-relative path") + if normalized in TEMP_WORKFLOW_PATHS: + raise SystemExit(f"{display_name} intended_files[{index}] must not include workflow handoff files") + if is_generated_path(normalized): + raise SystemExit(f"{display_name} intended_files[{index}] must not include generated/cache files") + return normalized + + +def validate_intended_files(metadata: dict[str, Any], *, display_name: str = "pr-metadata.json") -> list[str]: + raw_files = metadata.get(INTENDED_FILES_FIELD) + if not isinstance(raw_files, list) or not raw_files: + raise SystemExit(f"{display_name} field intended_files must be a non-empty list") + return [validate_intended_path(path, index, display_name=display_name) for index, path in enumerate(raw_files)] diff --git a/.github/scripts/prepare_issue_implementation_context.py b/.github/scripts/prepare_issue_implementation_context.py index 7e7f006..f5f4be1 100644 --- a/.github/scripts/prepare_issue_implementation_context.py +++ b/.github/scripts/prepare_issue_implementation_context.py @@ -4,7 +4,6 @@ from __future__ import annotations import argparse -import json import os import re import subprocess @@ -15,6 +14,20 @@ SCRIPT_DIR = Path(__file__).resolve().parent sys.path.insert(0, str(SCRIPT_DIR)) +from artifact_contracts import load_json, write_github_output, write_json # noqa: E402 +from context_snapshot import flatten_pages, format_issue_comments, remove_triggering_comment # noqa: E402 +from github_api import fetch_default_branch, run_gh_json # noqa: E402 +from github_event import ( # noqa: E402 + actor_login, + assignee_logins, + event_action, + event_assignee_login, + event_comment_body, + event_label_name, + is_pull_request_issue_event, + label_names, + triggering_comment_snapshot, +) from write_spec_context import ( # noqa: E402 APPROVED_LABEL, collect_spec_entries, @@ -30,36 +43,12 @@ IMPLEMENT_BRANCH_PREFIX = "spec/implement-issue" -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_event(path: str | None) -> dict[str, Any]: - if not path: - return {} - return json.loads(Path(path).read_text(encoding="utf-8")) + return load_json(path, default={}) def author_login(item: dict[str, Any]) -> str: - user = item.get("user") or item.get("author") or {} - return user.get("login") or "" - - -def label_names(issue: dict[str, Any]) -> list[str]: - return [label.get("name", "") for label in issue.get("labels", []) if label.get("name")] - - -def assignee_logins(issue: dict[str, Any]) -> list[str]: - return [assignee.get("login", "") for assignee in issue.get("assignees", []) if assignee.get("login")] + return actor_login(item) def comment_mentions_login(comment: str, login: str) -> bool: @@ -71,41 +60,8 @@ def comment_mentions_login(comment: str, login: str) -> bool: return bool(pattern.search(visible_comment)) -def event_comment_body(event: dict[str, Any]) -> str: - comment = event.get("comment") or {} - return comment.get("body") or "" - - -def is_pull_request_issue_event(event: dict[str, Any]) -> bool: - issue = event.get("issue") or {} - return bool(issue.get("pull_request")) - - -def event_action(event: dict[str, Any]) -> str: - return event.get("action") or "" - - -def event_label_name(event: dict[str, Any]) -> str: - label = event.get("label") or {} - return label.get("name") or "" - - -def event_assignee_login(event: dict[str, Any]) -> str: - assignee = event.get("assignee") or {} - return assignee.get("login") or "" - - def triggering_comment(event: dict[str, Any]) -> dict[str, Any] | None: - comment = event.get("comment") - if not comment: - return None - return { - "id": comment.get("id"), - "author": author_login(comment), - "body": comment.get("body") or "", - "created_at": comment.get("created_at") or "", - "url": comment.get("html_url") or "", - } + return triggering_comment_snapshot(event) def collect_coauthor_directives(*texts: str) -> list[str]: @@ -122,24 +78,6 @@ def collect_coauthor_directives(*texts: str) -> list[str]: return directives -def remove_triggering_comment( - comments: list[dict[str, Any]], - trigger_comment: dict[str, Any] | None, -) -> list[dict[str, Any]]: - if not trigger_comment: - return comments - trigger_id = trigger_comment.get("id") - trigger_url = trigger_comment.get("url") - filtered: list[dict[str, Any]] = [] - for comment in comments: - if trigger_id is not None and comment.get("id") == trigger_id: - continue - if trigger_url and comment.get("html_url") == trigger_url: - continue - filtered.append(comment) - return filtered - - def fetch_issue(repo: str, issue_number: int) -> dict[str, Any]: return run_gh_json( [ @@ -166,14 +104,6 @@ def fetch_comments(repo: str, issue_number: int) -> list[dict[str, Any]]: return flatten_pages(pages) -def fetch_default_branch(repo: str) -> str: - repository = run_gh_json(["repo", "view", repo, "--json", "defaultBranchRef"]) - default_branch = (repository.get("defaultBranchRef") or {}).get("name") - if not default_branch: - raise SystemExit("could not determine default branch") - return default_branch - - def extract_issue_number(args_issue: str, event: dict[str, Any]) -> int: if args_issue: return int(args_issue.lstrip("#")) @@ -292,28 +222,7 @@ def best_effort_assign(repo: str, issue_number: int, agent_login: str) -> str: def write_comments(path: Path, comments: list[dict[str, Any]]) -> None: - lines: list[str] = [] - for comment in comments: - lines.extend( - [ - f"Author: {author_login(comment)}", - f"Created: {comment.get('created_at') or ''}", - "", - comment.get("body") or "", - "", - "---", - "", - ] - ) - path.write_text("\n".join(lines), encoding="utf-8") - - -def write_github_output(path: str | None, values: dict[str, str]) -> None: - if not path: - return - with Path(path).open("a", encoding="utf-8") as handle: - for key, value in values.items(): - handle.write(f"{key}={value}\n") + path.write_text(format_issue_comments(comments), encoding="utf-8") def build_skipped_context(reason: str) -> dict[str, Any]: @@ -359,7 +268,7 @@ def build_skipped_context(reason: str) -> dict[str, Any]: def write_skipped_outputs(args: argparse.Namespace, reason: str) -> None: - Path(args.output).write_text(json.dumps(build_skipped_context(reason), indent=2) + "\n", encoding="utf-8") + write_json(args.output, build_skipped_context(reason), ensure_ascii=True) Path(args.comments_output).write_text("", encoding="utf-8") spec_output = Path(args.spec_context_output) if spec_output.exists(): @@ -381,27 +290,17 @@ def write_skipped_outputs(args: argparse.Namespace, reason: str) -> None: ) -def main() -> None: - parser = argparse.ArgumentParser() - parser.add_argument("--repo", required=True) - parser.add_argument("--issue", default="") - parser.add_argument("--event-path", default=os.environ.get("GITHUB_EVENT_PATH", "")) - parser.add_argument("--event-name", default=os.environ.get("GITHUB_EVENT_NAME", "")) - parser.add_argument("--agent-login", default="") - parser.add_argument("--output", default="issue_context.json") - parser.add_argument("--comments-output", default="issue_comments.txt") - parser.add_argument("--spec-context-output", default="spec_context.md") - parser.add_argument("--github-output", default=os.environ.get("GITHUB_OUTPUT", "")) - parser.add_argument("--force", action="store_true") - args = parser.parse_args() - - event = load_event(args.event_path) - - issue_number = extract_issue_number(args.issue, event) - issue = fetch_issue(args.repo, issue_number) - comments = fetch_comments(args.repo, issue_number) - default_branch = fetch_default_branch(args.repo) - run, reason = should_run(args, event, issue) +def build_implementation_context( + args: argparse.Namespace, + *, + event: dict[str, Any], + issue_number: int, + issue: dict[str, Any], + comments: list[dict[str, Any]], + default_branch: str, + should_run_flag: bool, + trigger_reason: str, +) -> tuple[dict[str, Any], list[dict[str, Any]], str]: trigger_comment = triggering_comment(event) historical_comments = remove_triggering_comment(comments, trigger_comment) spec_context = resolve_implementation_spec_context(args.repo, issue_number, default_branch) @@ -409,17 +308,13 @@ def main() -> None: selected_spec_pr = spec_context.get("selected_spec_pr") or {} target_branch = selected_spec_pr.get("head_ref_name") or implementation_target_branch(issue_number) noop = bool(spec_context.get("unapproved_spec_prs")) and not spec_context.get("spec_entries") - noop_reason = ( - "linked spec PR(s) exist for this issue but none are labeled plan-approved" - if noop - else "" - ) - assignment_warning = best_effort_assign(args.repo, issue_number, args.agent_login.strip()) if run else "" + noop_reason = "linked spec PR(s) exist for this issue but none are labeled plan-approved" if noop else "" + assignment_warning = best_effort_assign(args.repo, issue_number, args.agent_login.strip()) if should_run_flag else "" coauthor_directives = collect_coauthor_directives( issue.get("body") or "", *(comment.get("body") or "" for comment in comments), ) - existing_pr = has_existing_implementation_pr(args.repo, target_branch) if run and not selected_spec_pr else False + existing_pr = has_existing_implementation_pr(args.repo, target_branch) if should_run_flag and not selected_spec_pr else False context = { "owner": args.repo.split("/", 1)[0], @@ -453,15 +348,23 @@ def main() -> None: ".agents/skills/implement-issue/SKILL.md", ], "progress_start_line": f"Implementation run started for issue #{issue_number}.", - "should_run": run, + "should_run": should_run_flag, "should_noop": noop, - "skip_reason": "" if run else reason, + "skip_reason": "" if should_run_flag else trigger_reason, "noop_reason": noop_reason, - "trigger_reason": reason if run else "", + "trigger_reason": trigger_reason if should_run_flag else "", "assignment_warning": assignment_warning, } + return context, historical_comments, spec_context_text + - Path(args.output).write_text(json.dumps(context, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") +def write_context_outputs( + args: argparse.Namespace, + context: dict[str, Any], + historical_comments: list[dict[str, Any]], + spec_context_text: str, +) -> None: + write_json(args.output, context) write_comments(Path(args.comments_output), historical_comments) spec_output = Path(args.spec_context_output) if spec_context_text: @@ -471,19 +374,58 @@ def main() -> None: write_github_output( args.github_output, { - "should_run": "true" if run else "false", - "should_noop": "true" if noop else "false", - "skip_reason": "" if run else reason, - "noop_reason": noop_reason, - "issue_number": str(issue_number), - "default_branch": default_branch, - "target_branch": target_branch, + "should_run": "true" if context["should_run"] else "false", + "should_noop": "true" if context["should_noop"] else "false", + "skip_reason": str(context["skip_reason"]), + "noop_reason": str(context["noop_reason"]), + "issue_number": str(context["issue_number"]), + "default_branch": str(context["default_branch"]), + "target_branch": str(context["target_branch"]), "spec_context_source": str(context["spec_context_source"]), "selected_spec_pr_number": str(context["selected_spec_pr_number"] or ""), - "has_existing_implementation_pr": "true" if existing_pr else "false", + "has_existing_implementation_pr": "true" if context["has_existing_implementation_pr"] else "false", }, ) +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser() + parser.add_argument("--repo", required=True) + parser.add_argument("--issue", default="") + parser.add_argument("--event-path", default=os.environ.get("GITHUB_EVENT_PATH", "")) + parser.add_argument("--event-name", default=os.environ.get("GITHUB_EVENT_NAME", "")) + parser.add_argument("--agent-login", default="") + parser.add_argument("--output", default="issue_context.json") + parser.add_argument("--comments-output", default="issue_comments.txt") + parser.add_argument("--spec-context-output", default="spec_context.md") + parser.add_argument("--github-output", default=os.environ.get("GITHUB_OUTPUT", "")) + parser.add_argument("--force", action="store_true") + return parser.parse_args(argv) + + +def run(args: argparse.Namespace) -> None: + event = load_event(args.event_path) + issue_number = extract_issue_number(args.issue, event) + issue = fetch_issue(args.repo, issue_number) + comments = fetch_comments(args.repo, issue_number) + default_branch = fetch_default_branch(args.repo) + should_run_flag, trigger_reason = should_run(args, event, issue) + context, historical_comments, spec_context_text = build_implementation_context( + args, + event=event, + issue_number=issue_number, + issue=issue, + comments=comments, + default_branch=default_branch, + should_run_flag=should_run_flag, + trigger_reason=trigger_reason, + ) + write_context_outputs(args, context, historical_comments, spec_context_text) + + +def main() -> None: + run(parse_args()) + + if __name__ == "__main__": main() diff --git a/.github/scripts/prepare_issue_spec_context.py b/.github/scripts/prepare_issue_spec_context.py index bf8f337..32dbcae 100644 --- a/.github/scripts/prepare_issue_spec_context.py +++ b/.github/scripts/prepare_issue_spec_context.py @@ -4,23 +4,25 @@ from __future__ import annotations import argparse -import json import os import re -import subprocess from pathlib import Path from typing import Any - -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) - +from artifact_contracts import load_json, write_github_output, write_json +from context_snapshot import flatten_pages, format_issue_comments, remove_triggering_comment +from github_api import fetch_default_branch, run_gh_json +from github_event import ( + actor_login, + assignee_logins, + event_action, + event_assignee_login, + event_comment_body as event_comment_body_from_event, + event_label_name, + is_pull_request_issue_event, + label_names, + triggering_comment_snapshot, +) def spec_paths(issue_number: int) -> dict[str, str]: spec_dir = f"specs/issue-{issue_number}" @@ -39,7 +41,7 @@ def extract_issue_number(args_issue: str, event_path: str | None) -> int: return int(args_issue.lstrip("#")) if not event_path: raise SystemExit("--issue or --event-path is required") - event = json.loads(Path(event_path).read_text(encoding="utf-8")) + event = load_event(event_path) issue = event.get("issue") if issue and issue.get("number"): return int(issue["number"]) @@ -47,8 +49,7 @@ def extract_issue_number(args_issue: str, event_path: str | None) -> int: def author_login(item: dict[str, Any]) -> str: - user = item.get("user") or {} - return user.get("login") or "" + return actor_login(item) def fetch_issue(repo: str, issue_number: int) -> dict[str, Any]: @@ -75,58 +76,15 @@ def fetch_comments(repo: str, issue_number: int) -> list[dict[str, Any]]: "--slurp", ] ) - if pages and all(isinstance(page, list) for page in pages): - return [comment for page in pages for comment in page] - if isinstance(pages, list): - return pages - raise SystemExit("unexpected gh api comments response") - - -def fetch_default_branch(repo: str) -> str: - repository = run_gh_json(["repo", "view", repo, "--json", "defaultBranchRef"]) - default_branch = (repository.get("defaultBranchRef") or {}).get("name") - if not default_branch: - raise SystemExit("could not determine default branch") - return default_branch - - -def label_names(issue: dict[str, Any]) -> list[str]: - return [label.get("name", "") for label in issue.get("labels", []) if label.get("name")] - - -def assignee_logins(issue: dict[str, Any]) -> list[str]: - return [assignee.get("login", "") for assignee in issue.get("assignees", []) if assignee.get("login")] + return flatten_pages(pages) def load_event(event_path: str | None) -> dict[str, Any]: - if not event_path: - return {} - return json.loads(Path(event_path).read_text(encoding="utf-8")) - - -def event_action(event: dict[str, Any]) -> str: - return event.get("action") or "" - - -def event_label_name(event: dict[str, Any]) -> str: - label = event.get("label") or {} - return label.get("name") or "" - - -def event_assignee_login(event: dict[str, Any]) -> str: - assignee = event.get("assignee") or {} - return assignee.get("login") or "" + return load_json(event_path, default={}) def event_comment_body(event_path: str | None) -> str: - event = load_event(event_path) - comment = event.get("comment") or {} - return comment.get("body") or "" - - -def is_pull_request_issue_event(event: dict[str, Any]) -> bool: - issue = event.get("issue") or {} - return bool(issue.get("pull_request")) + return event_comment_body_from_event(load_event(event_path)) def comment_mentions_login(comment: str, login: str) -> bool: @@ -139,17 +97,7 @@ def comment_mentions_login(comment: str, login: str) -> bool: def triggering_comment(event_path: str | None) -> dict[str, Any] | None: - event = load_event(event_path) - comment = event.get("comment") - if not comment: - return None - return { - "id": comment.get("id"), - "author": author_login(comment), - "body": comment.get("body") or "", - "created_at": comment.get("created_at") or "", - "url": comment.get("html_url") or "", - } + return triggering_comment_snapshot(load_event(event_path)) def collect_coauthor_directives(*texts: str) -> list[str]: @@ -166,25 +114,6 @@ def collect_coauthor_directives(*texts: str) -> list[str]: return directives -def remove_triggering_comment( - comments: list[dict[str, Any]], - trigger_comment: dict[str, Any] | None, -) -> list[dict[str, Any]]: - if not trigger_comment: - return comments - - trigger_id = trigger_comment.get("id") - trigger_url = trigger_comment.get("url") - filtered: list[dict[str, Any]] = [] - for comment in comments: - if trigger_id is not None and comment.get("id") == trigger_id: - continue - if trigger_url and comment.get("html_url") == trigger_url: - continue - filtered.append(comment) - return filtered - - def should_run(args: argparse.Namespace, issue: dict[str, Any]) -> tuple[bool, str]: event = load_event(args.event_path) if args.event_name == "issue_comment" and is_pull_request_issue_event(event): @@ -230,31 +159,10 @@ def should_run(args: argparse.Namespace, issue: dict[str, Any]) -> tuple[bool, s def write_comments(path: Path, comments: list[dict[str, Any]]) -> None: - lines: list[str] = [] - for comment in comments: - lines.extend( - [ - f"Author: {author_login(comment)}", - f"Created: {comment.get('created_at') or ''}", - "", - comment.get("body") or "", - "", - "---", - "", - ] - ) - path.write_text("\n".join(lines), encoding="utf-8") - - -def write_github_output(path: str | None, values: dict[str, str]) -> None: - if not path: - return - with Path(path).open("a", encoding="utf-8") as handle: - for key, value in values.items(): - handle.write(f"{key}={value}\n") + path.write_text(format_issue_comments(comments), encoding="utf-8") -def main() -> None: +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument("--repo", required=True) parser.add_argument("--issue", default="") @@ -265,8 +173,10 @@ def main() -> None: parser.add_argument("--comments-output", default="issue_comments.txt") parser.add_argument("--github-output", default=os.environ.get("GITHUB_OUTPUT", "")) parser.add_argument("--force", action="store_true") - args = parser.parse_args() + return parser.parse_args(argv) + +def run(args: argparse.Namespace) -> None: issue_number = extract_issue_number(args.issue, args.event_path) issue = fetch_issue(args.repo, issue_number) comments = fetch_comments(args.repo, issue_number) @@ -293,7 +203,7 @@ def main() -> None: "trigger_reason": reason if run else "", } - Path(args.output).write_text(json.dumps(context, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + write_json(args.output, context) write_comments(Path(args.comments_output), historical_comments) write_github_output( args.github_output, @@ -311,5 +221,9 @@ def main() -> None: ) +def main() -> None: + run(parse_args()) + + if __name__ == "__main__": main() diff --git a/.github/scripts/prepare_issue_triage_context.py b/.github/scripts/prepare_issue_triage_context.py index a99bc4e..c250f6a 100644 --- a/.github/scripts/prepare_issue_triage_context.py +++ b/.github/scripts/prepare_issue_triage_context.py @@ -4,35 +4,31 @@ from __future__ import annotations import argparse -import json import os -import subprocess from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any +from artifact_contracts import load_json, write_github_output, write_json +from context_snapshot import flatten_pages, remove_triggering_comment +from github_api import fetch_default_branch, run_gh_json +from github_event import ( + actor_login, + assignee_logins, + event_comment, + event_issue, + is_pull_request_issue_event, + label_names, + triggering_comment_snapshot, +) + TRUSTED_COMMENT_ASSOCIATIONS = {"OWNER", "MEMBER", "COLLABORATOR"} NEEDS_INFO_LABEL = "needs-info" -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_event(path: str | None) -> dict[str, Any]: - if not path: - return {} - return json.loads(Path(path).read_text(encoding="utf-8")) + return load_json(path, default={}) def extract_issue_number(args_issue: str, event: dict[str, Any]) -> int: @@ -45,8 +41,7 @@ def extract_issue_number(args_issue: str, event: dict[str, Any]) -> int: def author_login(item: dict[str, Any]) -> str: - user = item.get("user") or item.get("author") or {} - return user.get("login") or "" + return actor_login(item) def is_automation_user(item: dict[str, Any]) -> bool: @@ -56,29 +51,6 @@ def is_automation_user(item: dict[str, Any]) -> bool: return user_type == "bot" or login.endswith("[bot]") -def label_names(issue: dict[str, Any]) -> list[str]: - return [label.get("name", "") for label in issue.get("labels", []) if label.get("name")] - - -def assignee_logins(issue: dict[str, Any]) -> list[str]: - return [assignee.get("login", "") for assignee in issue.get("assignees", []) if assignee.get("login")] - - -def is_pull_request_issue_event(event: dict[str, Any]) -> bool: - issue = event.get("issue") or {} - return "pull_request" in issue and issue.get("pull_request") is not None - - -def event_issue(event: dict[str, Any]) -> dict[str, Any]: - issue = event.get("issue") - return issue if isinstance(issue, dict) else {} - - -def event_comment(event: dict[str, Any]) -> dict[str, Any]: - comment = event.get("comment") - return comment if isinstance(comment, dict) else {} - - def issue_has_label(issue: dict[str, Any], label: str) -> bool: return label in label_names(issue) @@ -106,35 +78,7 @@ def comment_has_triage_command(comment: object, login: str) -> bool: def triggering_comment(event: dict[str, Any]) -> dict[str, Any] | None: - comment = event_comment(event) - if not comment: - return None - return { - "id": comment.get("id"), - "author": author_login(comment), - "author_association": comment.get("author_association") or "", - "body": comment.get("body") or "", - "created_at": comment.get("created_at") or "", - "url": comment.get("html_url") or "", - } - - -def remove_triggering_comment( - comments: list[dict[str, Any]], - trigger_comment: dict[str, Any] | None, -) -> list[dict[str, Any]]: - if not trigger_comment: - return comments - trigger_id = trigger_comment.get("id") - trigger_url = trigger_comment.get("url") - filtered: list[dict[str, Any]] = [] - for comment in comments: - if trigger_id is not None and comment.get("id") == trigger_id: - continue - if trigger_url and comment.get("html_url") == trigger_url: - continue - filtered.append(comment) - return filtered + return triggering_comment_snapshot(event, include_author_association=True) def should_run(args: argparse.Namespace, event: dict[str, Any]) -> tuple[bool, str]: @@ -282,18 +226,10 @@ def dedupe_candidates(repo: str, current_issue_number: int, *, now: datetime | N return candidates -def fetch_default_branch(repo: str) -> str: - repository = run_gh_json(["repo", "view", repo, "--json", "defaultBranchRef"]) - default_branch = (repository.get("defaultBranchRef") or {}).get("name") - if not default_branch: - raise SystemExit("could not determine default branch") - return default_branch - - def load_config(path: Path) -> dict[str, Any]: if not path.exists(): return {"labels": {}} - return json.loads(path.read_text(encoding="utf-8")) + return load_json(path, default={"labels": {}}) def read_issue_templates(root: Path) -> list[dict[str, str]]: @@ -347,45 +283,26 @@ def write_templates(path: Path, templates: list[dict[str, str]]) -> None: def write_dedupe_candidates(path: Path, candidates: list[dict[str, Any]]) -> None: - path.write_text(json.dumps(candidates, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + write_json(path, candidates) -def write_github_output(path: str | None, values: dict[str, str]) -> None: - if not path: - return - with Path(path).open("a", encoding="utf-8") as handle: - for key, value in values.items(): - handle.write(f"{key}={value}\n") - - -def main() -> None: - parser = argparse.ArgumentParser() - parser.add_argument("--repo", required=True) - parser.add_argument("--issue", default="") - parser.add_argument("--event-path", default=os.environ.get("GITHUB_EVENT_PATH", "")) - parser.add_argument("--event-name", default=os.environ.get("GITHUB_EVENT_NAME", "")) - parser.add_argument("--agent-login", default="") - parser.add_argument("--include-issue-body", action="store_true") - parser.add_argument("--output", default="triage_context.json") - parser.add_argument("--comments-output", default="issue_comments.txt") - parser.add_argument("--templates-output", default="issue_templates.txt") - parser.add_argument("--dedupe-output", default="dedupe_candidates.json") - parser.add_argument("--github-output", default=os.environ.get("GITHUB_OUTPUT", "")) - args = parser.parse_args() - - event = load_event(args.event_path) - issue_number = extract_issue_number(args.issue, event) - issue = fetch_issue(args.repo, issue_number) - comments = fetch_comments(args.repo, issue_number) - candidates = dedupe_candidates(args.repo, issue_number) - default_branch = fetch_default_branch(args.repo) - run, reason = should_run(args, event) +def build_triage_context( + args: argparse.Namespace, + *, + event: dict[str, Any], + issue_number: int, + issue: dict[str, Any], + comments: list[dict[str, Any]], + candidates: list[dict[str, Any]], + default_branch: str, + should_run_flag: bool, + trigger_reason: str, + root: Path, + config: dict[str, Any], + templates: list[dict[str, str]], +) -> tuple[dict[str, Any], list[dict[str, Any]]]: trigger_comment = triggering_comment(event) if args.event_name == "issue_comment" else None historical_comments = remove_triggering_comment(comments, trigger_comment) - root = Path.cwd() - config = load_config(root / ".github" / "issue-triage" / "config.json") - templates = read_issue_templates(root) - context = { "owner": args.repo.split("/", 1)[0], "repo": args.repo.split("/", 1)[1] if "/" in args.repo else args.repo, @@ -406,7 +323,7 @@ def main() -> None: "comments_count": len(comments), "historical_comments_count": len(historical_comments), "triggering_comment": trigger_comment, - "trigger_reason": reason if run else "", + "trigger_reason": trigger_reason if should_run_flag else "", "triage_config": config, "issue_template_paths": [template["path"] for template in templates], "dedupe_candidates_path": args.dedupe_output, @@ -423,25 +340,82 @@ def main() -> None: ".agents/skills/triage-issue-local/SKILL.md", ".agents/skills/dedupe-issue-local/SKILL.md", ], - "should_run": run, - "skip_reason": "" if run else reason, + "should_run": should_run_flag, + "skip_reason": "" if should_run_flag else trigger_reason, } + return context, historical_comments + - Path(args.output).write_text(json.dumps(context, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") +def write_triage_outputs( + args: argparse.Namespace, + context: dict[str, Any], + historical_comments: list[dict[str, Any]], + templates: list[dict[str, str]], + candidates: list[dict[str, Any]], +) -> None: + write_json(args.output, context) write_comments(Path(args.comments_output), historical_comments) write_templates(Path(args.templates_output), templates) write_dedupe_candidates(Path(args.dedupe_output), candidates) write_github_output( args.github_output, { - "should_run": "true" if run else "false", - "skip_reason": "" if run else reason, - "issue_number": str(issue_number), - "default_branch": default_branch, - "include_issue_body": "true" if args.include_issue_body else "false", + "should_run": "true" if context["should_run"] else "false", + "skip_reason": str(context["skip_reason"]), + "issue_number": str(context["issue_number"]), + "default_branch": str(context["default_branch"]), + "include_issue_body": "true" if context["include_issue_body"] else "false", }, ) +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser() + parser.add_argument("--repo", required=True) + parser.add_argument("--issue", default="") + parser.add_argument("--event-path", default=os.environ.get("GITHUB_EVENT_PATH", "")) + parser.add_argument("--event-name", default=os.environ.get("GITHUB_EVENT_NAME", "")) + parser.add_argument("--agent-login", default="") + parser.add_argument("--include-issue-body", action="store_true") + parser.add_argument("--output", default="triage_context.json") + parser.add_argument("--comments-output", default="issue_comments.txt") + parser.add_argument("--templates-output", default="issue_templates.txt") + parser.add_argument("--dedupe-output", default="dedupe_candidates.json") + parser.add_argument("--github-output", default=os.environ.get("GITHUB_OUTPUT", "")) + return parser.parse_args(argv) + + +def run(args: argparse.Namespace) -> None: + event = load_event(args.event_path) + issue_number = extract_issue_number(args.issue, event) + issue = fetch_issue(args.repo, issue_number) + comments = fetch_comments(args.repo, issue_number) + candidates = dedupe_candidates(args.repo, issue_number) + default_branch = fetch_default_branch(args.repo) + should_run_flag, trigger_reason = should_run(args, event) + root = Path.cwd() + config = load_config(root / ".github" / "issue-triage" / "config.json") + templates = read_issue_templates(root) + context, historical_comments = build_triage_context( + args, + event=event, + issue_number=issue_number, + issue=issue, + comments=comments, + candidates=candidates, + default_branch=default_branch, + should_run_flag=should_run_flag, + trigger_reason=trigger_reason, + root=root, + config=config, + templates=templates, + ) + write_triage_outputs(args, context, historical_comments, templates, candidates) + + +def main() -> None: + run(parse_args()) + + if __name__ == "__main__": main() diff --git a/.github/scripts/prepare_local_review_inputs.py b/.github/scripts/prepare_local_review_inputs.py index 8bceb7f..e87a345 100644 --- a/.github/scripts/prepare_local_review_inputs.py +++ b/.github/scripts/prepare_local_review_inputs.py @@ -16,6 +16,7 @@ SCRIPT_DIR = Path(__file__).resolve().parent sys.path.insert(0, str(SCRIPT_DIR)) +from artifact_contracts import write_github_output # noqa: E402 import build_pr_diff # noqa: E402 import select_review_skill # noqa: E402 import write_pr_description # noqa: E402 @@ -34,14 +35,14 @@ StatusRecord = tuple[str, str, str, str] -def run(args: list[str], env: dict[str, str] | None = None) -> str: +def run_command(args: list[str], env: dict[str, str] | None = None) -> str: result = subprocess.run(args, check=True, stdout=subprocess.PIPE, text=True, env=env) return result.stdout.strip() def optional_git(args: list[str]) -> str: try: - return run(["git", *args]) + return run_command(["git", *args]) except subprocess.CalledProcessError: return "" @@ -107,7 +108,7 @@ def filtered_status_records() -> list[StatusRecord]: def resolve_ref(ref: str) -> str: - return run(["git", "rev-parse", ref]) + return run_command(["git", "rev-parse", ref]) def default_base() -> str: @@ -196,7 +197,7 @@ def github_pr_event(repo: str, branch: str) -> dict[str, Any]: "headRefOid", ] ) - data = json.loads(run(["gh", "pr", "view", branch, "--repo", repo, "--json", fields])) + data = json.loads(run_command(["gh", "pr", "view", branch, "--repo", repo, "--json", fields])) return { "pull_request": { "number": data.get("number") or "", @@ -253,7 +254,7 @@ def sync_event_base(event: dict[str, Any], base: str, base_sha: str) -> None: def run_git_with_index(args: list[str], index_path: str) -> str: env = os.environ.copy() env["GIT_INDEX_FILE"] = index_path - return run(["git", *args], env=env) + return run_command(["git", *args], env=env) def local_worktree_diff(base_sha: str, context_lines: int) -> list[str]: @@ -319,22 +320,16 @@ def write_spec_context_if_needed(repo: str, event: dict[str, Any], pr_diff_text: output.unlink() -def write_github_output(path: str | None, values: dict[str, str]) -> None: - if not path: - return - with Path(path).open("a", encoding="utf-8") as handle: - for key, value in values.items(): - handle.write(f"{key}={value}\n") - - -def main() -> int: +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument("--repo", default="") parser.add_argument("--base", default="") parser.add_argument("--head", default="HEAD") parser.add_argument("--github-output", default=os.environ.get("GITHUB_OUTPUT", "")) - args = parser.parse_args() + return parser.parse_args(argv) + +def run(args: argparse.Namespace) -> int: remove_stale_review_files() repo = args.repo or default_repo() @@ -380,5 +375,9 @@ def main() -> int: return 0 +def main() -> int: + return run(parse_args()) + + if __name__ == "__main__": raise SystemExit(main()) diff --git a/.github/scripts/prepare_pr_comment_context.py b/.github/scripts/prepare_pr_comment_context.py index d22ec22..9c9e46c 100644 --- a/.github/scripts/prepare_pr_comment_context.py +++ b/.github/scripts/prepare_pr_comment_context.py @@ -14,6 +14,8 @@ SCRIPT_DIR = Path(__file__).resolve().parent sys.path.insert(0, str(SCRIPT_DIR)) +from artifact_contracts import write_github_output # noqa: E402 +from github_api import fetch_default_branch, flatten_gh_pages as flatten_pages, run_gh_json # noqa: E402 from prepare_issue_implementation_context import collect_coauthor_directives # noqa: E402 from resolve_pr_event import comment_has_fix_command # noqa: E402 @@ -25,19 +27,6 @@ PAGE_INFO = "pageInfo { hasNextPage endCursor }" -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_event(path: str | None) -> dict[str, Any]: if not path: return {} @@ -57,14 +46,6 @@ def fetch_pr(repo: str, number: int) -> dict[str, Any]: return run_gh_json(["api", f"repos/{repo}/pulls/{number}"]) -def fetch_default_branch(repo: str) -> str: - repository = run_gh_json(["repo", "view", repo, "--json", "defaultBranchRef"]) - default_branch = (repository.get("defaultBranchRef") or {}).get("name") - if not default_branch: - raise SystemExit("could not determine default branch") - return default_branch - - def fetch_collaborator_permission(repo: str, login: str) -> str: if not login: return "" @@ -386,36 +367,47 @@ def review_comment_index(comments: list[dict[str, Any]], threads: list[dict[str, return {"review_comments": items} -def build_context(repo: str, event_name: str, event: dict[str, Any], agent_login: str) -> tuple[dict[str, Any], dict[str, Any]]: - trigger = fill_trigger_author_metadata(repo, event_trigger(event_name, event)) - pr = fetch_pr(repo, trigger["pr_number"]) - default_branch = fetch_default_branch(repo) - has_command = comment_has_fix_command(trigger["body"], agent_login) - auth = trigger_authorization(repo, pr, trigger) +def run_decision( + *, + agent_login: str, + has_command: bool, + auth: dict[str, Any], + pr_state: str, + strategy: dict[str, Any], +) -> tuple[bool, str]: authorized = bool(auth["authorized"]) - state = str(pr.get("state") or "").lower() - strategy = branch_strategy(repo, pr, authorized) - - should_run = has_command and authorized and state == "open" and strategy["branch_strategy"] != "blocked" + should_run = has_command and authorized and pr_state == "open" and strategy["branch_strategy"] != "blocked" if not agent_login: - skip_reason = "agent login is not configured" - elif not has_command: - skip_reason = "missing valid @AGENT_LOGIN /fix command" - elif not authorized: - skip_reason = str(auth["skip_reason"]) - elif state != "open": - skip_reason = f"pull request is {state or 'not open'}" - elif strategy["branch_strategy"] == "blocked": - skip_reason = "no writable branch strategy is available" - else: - skip_reason = "" - + return should_run, "agent login is not configured" + if not has_command: + return should_run, "missing valid @AGENT_LOGIN /fix command" + if not authorized: + return should_run, str(auth["skip_reason"]) + if pr_state != "open": + return should_run, f"pull request is {pr_state or 'not open'}" + if strategy["branch_strategy"] == "blocked": + return should_run, "no writable branch strategy is available" + return should_run, "" + + +def build_pr_comment_context_payload( + repo: str, + *, + trigger: dict[str, Any], + pr: dict[str, Any], + default_branch: str, + strategy: dict[str, Any], + auth: dict[str, Any], + has_command: bool, + should_run: bool, + skip_reason: str, +) -> dict[str, Any]: head = pr.get("head") or {} base = pr.get("base") or {} head_repo = (head.get("repo") or {}).get("full_name") or "" base_repo = (base.get("repo") or {}).get("full_name") or repo owner, name = repo.split("/", 1) - context = { + return { "owner": owner, "repo": name, "repository": repo, @@ -434,7 +426,7 @@ def build_context(repo: str, event_name: str, event: dict[str, Any], agent_login **strategy, **{key: value for key, value in trigger.items() if key != "body"}, "trigger_body": trigger["body"], - "trigger_actor_is_authorized": authorized, + "trigger_actor_is_authorized": bool(auth["authorized"]), "trigger_actor_repository_permission": auth["permission"], "base_repo_private": auth["private_repo"], "trigger_command_present": has_command, @@ -449,18 +441,39 @@ def build_context(repo: str, event_name: str, event: dict[str, Any], agent_login "should_noop": False, "skip_reason": skip_reason, } - return context, pr -def write_github_output(path: str | None, values: dict[str, str]) -> None: - if not path: - return - with Path(path).open("a", encoding="utf-8") as handle: - for key, value in values.items(): - handle.write(f"{key}={value}\n") +def build_context(repo: str, event_name: str, event: dict[str, Any], agent_login: str) -> tuple[dict[str, Any], dict[str, Any]]: + trigger = fill_trigger_author_metadata(repo, event_trigger(event_name, event)) + pr = fetch_pr(repo, trigger["pr_number"]) + default_branch = fetch_default_branch(repo) + has_command = comment_has_fix_command(trigger["body"], agent_login) + auth = trigger_authorization(repo, pr, trigger) + authorized = bool(auth["authorized"]) + state = str(pr.get("state") or "").lower() + strategy = branch_strategy(repo, pr, authorized) + should_run, skip_reason = run_decision( + agent_login=agent_login, + has_command=has_command, + auth=auth, + pr_state=state, + strategy=strategy, + ) + context = build_pr_comment_context_payload( + repo, + trigger=trigger, + pr=pr, + default_branch=default_branch, + strategy=strategy, + auth=auth, + has_command=has_command, + should_run=should_run, + skip_reason=skip_reason, + ) + return context, pr -def main() -> None: +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument("--repo", required=True) parser.add_argument("--event-path", default=os.environ.get("GITHUB_EVENT_PATH", "")) @@ -470,8 +483,10 @@ def main() -> None: parser.add_argument("--pr-event-output", default="pr_event.json") parser.add_argument("--review-comment-ids-output", default="review_comment_ids.json") parser.add_argument("--github-output", default=os.environ.get("GITHUB_OUTPUT", "")) - args = parser.parse_args() + return parser.parse_args(argv) + +def run(args: argparse.Namespace) -> None: event = load_event(args.event_path) context, pr = build_context(args.repo, args.event_name, event, args.agent_login.strip()) comments = fetch_review_comments(args.repo, int(context["pr_number"])) @@ -500,5 +515,9 @@ def main() -> None: ) +def main() -> None: + run(parse_args()) + + if __name__ == "__main__": main() diff --git a/.github/scripts/prepare_product_change_report_context.py b/.github/scripts/prepare_product_change_report_context.py index e8fde84..51bca3e 100644 --- a/.github/scripts/prepare_product_change_report_context.py +++ b/.github/scripts/prepare_product_change_report_context.py @@ -6,25 +6,17 @@ import argparse import datetime as dt import json -import subprocess from pathlib import Path from typing import Any +from artifact_contracts import write_github_output +from github_api import fetch_default_branch, run_gh_json, run_gh_text +from ledger_contracts import entries_by_pr, load_ledger as load_pr_ledger UTC = dt.timezone.utc DEFAULT_LEDGER_PATH = "docs/updates/.product-change-report-ledger.json" -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 run_gh_text(args: list[str]) -> str: - result = subprocess.run(["gh", *args], check=True, stdout=subprocess.PIPE, text=True) - return result.stdout - - def parse_date(value: str) -> dt.date: return dt.date.fromisoformat(value) @@ -84,16 +76,8 @@ def iso_z(value: dt.datetime) -> str: return value.astimezone(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z") -def fetch_default_branch(repo: str) -> str: - data = run_gh_json(["repo", "view", repo, "--json", "defaultBranchRef"]) - branch = (data.get("defaultBranchRef") or {}).get("name") - if not branch: - raise SystemExit("could not determine default branch") - return branch - - def search_merged_pr_numbers(repo: str, start: dt.datetime, end: dt.datetime) -> list[int]: - branch = fetch_default_branch(repo) + branch = fetch_default_branch(repo, run_json=run_gh_json) numbers: list[int] = [] seen: set[int] = set() current_date = start.date() @@ -212,29 +196,11 @@ def issue_refs(pr: dict[str, Any]) -> list[str]: def load_ledger(path: Path) -> dict[str, Any]: - if not path.exists(): - return {"version": 1, "entries": []} - data = json.loads(path.read_text(encoding="utf-8")) - if not isinstance(data, dict): - raise SystemExit(f"invalid product change report ledger: {path}") - entries = data.get("entries") - if entries is None: - data["entries"] = [] - elif not isinstance(entries, list): - raise SystemExit(f"invalid product change report ledger entries: {path}") - data.setdefault("version", 1) - return data + return load_pr_ledger(path, ledger_name="product change report") def ledger_entries_by_pr(ledger: dict[str, Any]) -> dict[int, dict[str, Any]]: - entries: dict[int, dict[str, Any]] = {} - for entry in ledger.get("entries") or []: - try: - pr_number = int(entry.get("pr")) - except (TypeError, ValueError): - continue - entries[pr_number] = entry - return entries + return entries_by_pr(ledger) def split_prs_by_ledger( @@ -358,15 +324,7 @@ def write_diffs(path: Path, repo: str, prs: list[dict[str, Any]], max_chars_per_ path.write_text("\n".join(lines), encoding="utf-8") -def write_github_output(path: str | None, values: dict[str, str]) -> None: - if not path: - return - with Path(path).open("a", encoding="utf-8") as handle: - for key, value in values.items(): - handle.write(f"{key}={value}\n") - - -def main() -> int: +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument("--repo", required=True) parser.add_argument("--report-date", default="") @@ -378,8 +336,10 @@ def main() -> int: parser.add_argument("--github-output", default="") parser.add_argument("--ledger-path", default=DEFAULT_LEDGER_PATH) parser.add_argument("--max-diff-chars-per-pr", type=int, default=60000) - args = parser.parse_args() + return parser.parse_args(argv) + +def run(args: argparse.Namespace) -> int: report_id, start, end = resolve_scan_window(args.report_date, args.start_date, args.end_date) default_branch = fetch_default_branch(args.repo) report_path = report_path_for_id(report_id) @@ -415,5 +375,9 @@ def main() -> int: return 0 +def main() -> int: + return run(parse_args()) + + if __name__ == "__main__": raise SystemExit(main()) diff --git a/.github/scripts/prepare_product_docs_sync_context.py b/.github/scripts/prepare_product_docs_sync_context.py index 249857c..51e5691 100644 --- a/.github/scripts/prepare_product_docs_sync_context.py +++ b/.github/scripts/prepare_product_docs_sync_context.py @@ -11,6 +11,11 @@ from pathlib import Path from typing import Any +from artifact_contracts import write_github_output +from github_api import fetch_default_branch, run_gh_json, run_gh_text +from issue_refs import issue_numbers_from_closing_refs +from ledger_contracts import entries_by_pr, load_ledger as load_pr_ledger + UTC = dt.timezone.utc DEFAULT_LEDGER_PATH = "docs/product/.product-docs-sync-ledger.json" PRODUCT_DOCS_SYNC_BRANCH_PREFIX = "docs/product-docs-sync" @@ -20,32 +25,6 @@ "Record product docs sync decision for PR #", ) PRODUCT_DOCS_SYNC_TITLES = {"Update product docs"} -ISSUE_REFERENCE_KEYWORD_RE = re.compile( - r"\b(?:refs?|references?|relates\s+to|fixes?|closes?|resolves?)\b[:\s]+[^\n\r]*", - re.IGNORECASE, -) -ISSUE_NUMBER_RE = re.compile(r"#(\d+)\b") -PULL_REQUEST_REFERENCE_PREFIX_RE = re.compile(r"(?:\bPR|\bpull\s+request)\s*$", re.IGNORECASE) - - -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 run_gh_text(args: list[str]) -> str: - result = subprocess.run(["gh", *args], check=True, stdout=subprocess.PIPE, text=True) - return result.stdout - - -def fetch_default_branch(repo: str) -> str: - data = run_gh_json(["repo", "view", repo, "--json", "defaultBranchRef"]) - branch = (data.get("defaultBranchRef") or {}).get("name") - if not branch: - raise SystemExit("could not determine default branch") - return branch - - def parse_date(value: str) -> dt.date: return dt.date.fromisoformat(value) @@ -165,26 +144,11 @@ def fetch_merged_prs(repo: str, start: dt.datetime, end: dt.datetime, default_br def load_ledger(path: Path) -> dict[str, Any]: - if not path.exists(): - return {"version": 1, "entries": []} - data = json.loads(path.read_text(encoding="utf-8")) - if not isinstance(data, dict): - raise SystemExit(f"invalid product docs sync ledger: {path}") - data.setdefault("version", 1) - data.setdefault("entries", []) - if not isinstance(data["entries"], list): - raise SystemExit(f"invalid product docs sync ledger entries: {path}") - return data + return load_pr_ledger(path, ledger_name="product docs sync") def ledger_entries_by_pr(ledger: dict[str, Any]) -> dict[int, dict[str, Any]]: - entries: dict[int, dict[str, Any]] = {} - for entry in ledger.get("entries") or []: - try: - entries[int(entry.get("pr"))] = entry - except (TypeError, ValueError): - continue - return entries + return entries_by_pr(ledger) def select_unprocessed_pr(prs: list[dict[str, Any]], ledger: dict[str, Any]) -> tuple[dict[str, Any] | None, list[dict[str, Any]]]: @@ -242,26 +206,7 @@ def fetch_pr_diff(repo: str, pr_number: str, max_chars: int) -> str: def issue_numbers(pr: dict[str, Any]) -> list[int]: - numbers: list[int] = [] - - def add_number(value: Any) -> None: - try: - number = int(value) - except (TypeError, ValueError): - return - if number not in numbers: - numbers.append(number) - - for issue in pr.get("closingIssuesReferences") or []: - add_number(issue.get("number")) - for text in (pr.get("title") or "", pr.get("body") or ""): - for match in ISSUE_REFERENCE_KEYWORD_RE.finditer(str(text)): - reference_text = match.group(0) - for number_match in ISSUE_NUMBER_RE.finditer(reference_text): - if PULL_REQUEST_REFERENCE_PREFIX_RE.search(reference_text[: number_match.start()]): - continue - add_number(number_match.group(1)) - return numbers + return issue_numbers_from_closing_refs(pr) def compact_author(value: dict[str, Any] | None) -> str: @@ -400,15 +345,7 @@ def write_existing_docs(path: Path, product_docs: list[dict[str, str]]) -> None: path.write_text("\n".join(lines), encoding="utf-8") -def write_github_output(path: str | None, values: dict[str, str]) -> None: - if not path: - return - with Path(path).open("a", encoding="utf-8") as handle: - for key, value in values.items(): - handle.write(f"{key}={value}\n") - - -def main() -> int: +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument("--repo", required=True) parser.add_argument("--pr-number", default="") @@ -422,10 +359,100 @@ def main() -> int: parser.add_argument("--github-output", default="") parser.add_argument("--ledger-path", default=DEFAULT_LEDGER_PATH) parser.add_argument("--max-diff-chars", type=int, default=100000) - args = parser.parse_args() + return parser.parse_args(argv) - root = Path.cwd() - default_branch = fetch_default_branch(args.repo) + +def write_no_pr_outputs( + args: argparse.Namespace, + default_branch: str, + product_docs: list[dict[str, str]], + scanned_pr_count: int, + skipped_prs: list[dict[str, Any]], + scan_window_payload: dict[str, str] | None, +) -> None: + write_context_json( + Path(args.context_output), + args.repo, + default_branch, + {}, + [], + [], + product_docs, + args.ledger_path, + scanned_pr_count, + skipped_prs, + scan_window_payload, + ) + Path(args.markdown_output).write_text("No unprocessed merged PRs found.\n", encoding="utf-8") + Path(args.diff_output).write_text("", encoding="utf-8") + write_existing_docs(Path(args.existing_docs_output), product_docs) + write_github_output( + args.github_output, + { + "pr_number": "", + "pr_title": "", + "pr_url": "", + "merged_at": "", + "default_branch": default_branch, + "ledger_path": args.ledger_path, + "scanned_pr_count": str(scanned_pr_count), + "skipped_processed_pr_count": str(len(skipped_prs)), + "should_run": "false", + "skip_reason": "no unprocessed merged pull requests found", + }, + ) + + +def write_selected_pr_outputs( + args: argparse.Namespace, + default_branch: str, + pr: dict[str, Any], + issues: list[dict[str, Any]], + specs: list[dict[str, str]], + product_docs: list[dict[str, str]], + scanned_pr_count: int, + skipped_prs: list[dict[str, Any]], + scan_window_payload: dict[str, str] | None, +) -> None: + write_context_json( + Path(args.context_output), + args.repo, + default_branch, + pr, + issues, + specs, + product_docs, + args.ledger_path, + scanned_pr_count, + skipped_prs, + scan_window_payload, + ) + write_markdown(Path(args.markdown_output), pr, issues, specs) + selected_pr_number = str(pr.get("number") or args.pr_number) + Path(args.diff_output).write_text(fetch_pr_diff(args.repo, selected_pr_number, args.max_diff_chars), encoding="utf-8") + write_existing_docs(Path(args.existing_docs_output), product_docs) + should_run = "true" if pr.get("mergedAt") else "false" + write_github_output( + args.github_output, + { + "pr_number": str(pr.get("number") or args.pr_number), + "pr_title": str(pr.get("title") or ""), + "pr_url": str(pr.get("url") or ""), + "merged_at": str(pr.get("mergedAt") or ""), + "default_branch": default_branch, + "ledger_path": args.ledger_path, + "scanned_pr_count": str(scanned_pr_count), + "skipped_processed_pr_count": str(len(skipped_prs)), + "should_run": should_run, + "skip_reason": "" if should_run == "true" else "pull request is not merged", + }, + ) + + +def resolve_target_pr( + args: argparse.Namespace, + default_branch: str, +) -> tuple[dict[str, Any] | None, int, list[dict[str, Any]], dict[str, str] | None]: skipped_prs: list[dict[str, Any]] = [] scanned_pr_count = 1 scan_window_payload = None @@ -446,88 +473,50 @@ def main() -> int: scanned_prs = fetch_merged_prs(args.repo, start, end, default_branch) scanned_pr_count = len(scanned_prs) pr, skipped_prs = select_unprocessed_pr(scanned_prs, load_ledger(Path(args.ledger_path))) + return pr, scanned_pr_count, skipped_prs, scan_window_payload - if pr is None: - product_docs = read_existing_product_docs(root) - write_context_json( - Path(args.context_output), - args.repo, - default_branch, - {}, - [], - [], - product_docs, - args.ledger_path, - scanned_pr_count, - skipped_prs, - scan_window_payload, - ) - Path(args.markdown_output).write_text("No unprocessed merged PRs found.\n", encoding="utf-8") - Path(args.diff_output).write_text("", encoding="utf-8") - write_existing_docs(Path(args.existing_docs_output), product_docs) - write_github_output( - args.github_output, - { - "pr_number": "", - "pr_title": "", - "pr_url": "", - "merged_at": "", - "default_branch": default_branch, - "ledger_path": args.ledger_path, - "scanned_pr_count": str(scanned_pr_count), - "skipped_processed_pr_count": str(len(skipped_prs)), - "should_run": "false", - "skip_reason": "no unprocessed merged pull requests found", - }, - ) - return 0 +def load_linked_context(root: Path, repo: str, pr: dict[str, Any]) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: numbers = issue_numbers(pr) - issues, _skipped_issue_numbers = fetch_existing_issues(args.repo, numbers) + issues, _skipped_issue_numbers = fetch_existing_issues(repo, numbers) existing_issue_numbers = [] for issue in issues: try: existing_issue_numbers.append(int(issue.get("number"))) except (TypeError, ValueError): continue - specs = read_specs(root, existing_issue_numbers) - product_docs = read_existing_product_docs(root) - should_run = "true" if pr.get("mergedAt") else "false" + return issues, read_specs(root, existing_issue_numbers) - write_context_json( - Path(args.context_output), - args.repo, + +def run(args: argparse.Namespace) -> int: + root = Path.cwd() + default_branch = fetch_default_branch(args.repo, run_json=run_gh_json) + pr, scanned_pr_count, skipped_prs, scan_window_payload = resolve_target_pr(args, default_branch) + + if pr is None: + product_docs = read_existing_product_docs(root) + write_no_pr_outputs(args, default_branch, product_docs, scanned_pr_count, skipped_prs, scan_window_payload) + return 0 + + issues, specs = load_linked_context(root, args.repo, pr) + product_docs = read_existing_product_docs(root) + write_selected_pr_outputs( + args, default_branch, pr, issues, specs, product_docs, - args.ledger_path, scanned_pr_count, skipped_prs, scan_window_payload, ) - write_markdown(Path(args.markdown_output), pr, issues, specs) - selected_pr_number = str(pr.get("number") or args.pr_number) - Path(args.diff_output).write_text(fetch_pr_diff(args.repo, selected_pr_number, args.max_diff_chars), encoding="utf-8") - write_existing_docs(Path(args.existing_docs_output), product_docs) - write_github_output( - args.github_output, - { - "pr_number": str(pr.get("number") or args.pr_number), - "pr_title": str(pr.get("title") or ""), - "pr_url": str(pr.get("url") or ""), - "merged_at": str(pr.get("mergedAt") or ""), - "default_branch": default_branch, - "ledger_path": args.ledger_path, - "scanned_pr_count": str(scanned_pr_count), - "skipped_processed_pr_count": str(len(skipped_prs)), - "should_run": should_run, - "skip_reason": "" if should_run == "true" else "pull request is not merged", - }, - ) return 0 +def main() -> int: + return run(parse_args()) + + if __name__ == "__main__": raise SystemExit(main()) diff --git a/.github/scripts/read_branch_sha.py b/.github/scripts/read_branch_sha.py index 1ffc112..efa7ff0 100644 --- a/.github/scripts/read_branch_sha.py +++ b/.github/scripts/read_branch_sha.py @@ -9,6 +9,8 @@ from pathlib import Path from typing import Any +from artifact_contracts import write_github_output + def run_gh_json(args: list[str]) -> Any | None: try: @@ -102,14 +104,6 @@ def end_state(repo: str, branch_prefix: str, metadata_path: Path | None, snapsho } -def write_github_output(path: str | None, values: dict[str, str]) -> None: - if not path: - return - with Path(path).open("a", encoding="utf-8") as handle: - for key, value in values.items(): - handle.write(f"{key}={value}\n") - - def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--repo", required=True) diff --git a/.github/scripts/resolve_pr_event.py b/.github/scripts/resolve_pr_event.py index a6e7cba..1a54483 100644 --- a/.github/scripts/resolve_pr_event.py +++ b/.github/scripts/resolve_pr_event.py @@ -9,6 +9,8 @@ from pathlib import Path from typing import Any +from artifact_contracts import write_github_output + def load_json(path: Path) -> dict[str, Any]: return json.loads(path.read_text(encoding="utf-8")) @@ -138,14 +140,6 @@ def review_state(event: dict[str, Any], repo: str, event_name: str = "") -> dict } -def write_github_output(path: str | None, values: dict[str, str]) -> None: - if not path: - return - with Path(path).open("a", encoding="utf-8") as handle: - for key, value in values.items(): - handle.write(f"{key}={value}\n") - - def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--repo", required=True) diff --git a/.github/scripts/update_implementation_progress.py b/.github/scripts/update_implementation_progress.py index 33f8487..df6eae8 100644 --- a/.github/scripts/update_implementation_progress.py +++ b/.github/scripts/update_implementation_progress.py @@ -9,21 +9,11 @@ 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 = "" - - -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") +MARKER = "" def find_progress_comment(repo: str, issue_number: int) -> dict[str, Any] | None: diff --git a/.github/scripts/update_product_change_report_ledger.py b/.github/scripts/update_product_change_report_ledger.py index e196b40..110f594 100644 --- a/.github/scripts/update_product_change_report_ledger.py +++ b/.github/scripts/update_product_change_report_ledger.py @@ -4,39 +4,22 @@ from __future__ import annotations import argparse -import datetime as dt -import json from pathlib import Path from typing import Any - -UTC = dt.timezone.utc - - -def now_iso() -> str: - return dt.datetime.now(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z") - - -def load_json(path: Path) -> dict[str, Any]: - return json.loads(path.read_text(encoding="utf-8")) +from artifact_contracts import load_json +from ledger_contracts import ( + entries_by_pr, + load_ledger as load_pr_ledger, + merge_commit_oid, + now_iso, + set_sorted_entries, + write_ledger, +) def load_ledger(path: Path) -> dict[str, Any]: - if not path.exists(): - return {"version": 1, "entries": []} - data = load_json(path) - if not isinstance(data, dict): - raise SystemExit(f"invalid product change report ledger: {path}") - data.setdefault("version", 1) - data.setdefault("entries", []) - if not isinstance(data["entries"], list): - raise SystemExit(f"invalid product change report ledger entries: {path}") - return data - - -def merge_commit_oid(pr: dict[str, Any]) -> str: - merge_commit = pr.get("mergeCommit") or {} - return merge_commit.get("oid") or "" + return load_pr_ledger(path, ledger_name="product change report") def build_entry(pr: dict[str, Any], context: dict[str, Any], recorded_at: str, status: str) -> dict[str, Any]: @@ -54,12 +37,7 @@ def build_entry(pr: dict[str, Any], context: dict[str, Any], recorded_at: str, s def update_ledger(ledger: dict[str, Any], context: dict[str, Any], recorded_at: str, status: str = "reported") -> dict[str, Any]: - by_pr: dict[int, dict[str, Any]] = {} - for entry in ledger.get("entries") or []: - try: - by_pr[int(entry.get("pr"))] = entry - except (TypeError, ValueError): - continue + by_pr = entries_by_pr(ledger) for pr in context.get("reportable_prs") or []: pr_number = int(pr["number"]) @@ -69,9 +47,7 @@ def update_ledger(ledger: dict[str, Any], context: dict[str, Any], recorded_at: entry_recorded_at = str(existing["recorded_at"]) by_pr[pr_number] = build_entry(pr, context, entry_recorded_at, status) - ledger["version"] = 1 - ledger["entries"] = sorted(by_pr.values(), key=lambda item: (item.get("merged_at") or "", int(item.get("pr") or 0))) - return ledger + return set_sorted_entries(ledger, list(by_pr.values())) def main() -> int: @@ -89,7 +65,7 @@ def main() -> int: ledger_path = Path(args.ledger or context.get("ledger_path") or "docs/updates/.product-change-report-ledger.json") ledger_path.parent.mkdir(parents=True, exist_ok=True) ledger = update_ledger(load_ledger(ledger_path), context, now_iso(), args.status) - ledger_path.write_text(json.dumps(ledger, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + write_ledger(ledger_path, ledger) return 0 diff --git a/.github/scripts/update_product_docs_sync_ledger.py b/.github/scripts/update_product_docs_sync_ledger.py index d1e9a0c..230f9fe 100644 --- a/.github/scripts/update_product_docs_sync_ledger.py +++ b/.github/scripts/update_product_docs_sync_ledger.py @@ -4,40 +4,24 @@ from __future__ import annotations import argparse -import datetime as dt -import json from pathlib import Path from typing import Any +from artifact_contracts import load_json +from ledger_contracts import ( + entries_by_pr, + load_ledger as load_pr_ledger, + merge_commit_oid, + now_iso, + set_sorted_entries, + write_ledger, +) -UTC = dt.timezone.utc DEFAULT_LEDGER_PATH = "docs/product/.product-docs-sync-ledger.json" -def now_iso() -> str: - return dt.datetime.now(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z") - - -def load_json(path: Path) -> dict[str, Any]: - return json.loads(path.read_text(encoding="utf-8")) - - def load_ledger(path: Path) -> dict[str, Any]: - if not path.exists(): - return {"version": 1, "entries": []} - data = load_json(path) - if not isinstance(data, dict): - raise SystemExit(f"invalid product docs sync ledger: {path}") - data.setdefault("version", 1) - data.setdefault("entries", []) - if not isinstance(data["entries"], list): - raise SystemExit(f"invalid product docs sync ledger entries: {path}") - return data - - -def merge_commit_oid(pr: dict[str, Any]) -> str: - merge_commit = pr.get("mergeCommit") or {} - return merge_commit.get("oid") or "" + return load_pr_ledger(path, ledger_name="product docs sync") def build_entry(context: dict[str, Any], result: dict[str, Any], recorded_at: str) -> dict[str, Any]: @@ -63,12 +47,7 @@ def update_ledger( result: dict[str, Any], recorded_at: str, ) -> dict[str, Any]: - by_pr: dict[int, dict[str, Any]] = {} - for entry in ledger.get("entries") or []: - try: - by_pr[int(entry.get("pr"))] = entry - except (TypeError, ValueError): - continue + by_pr = entries_by_pr(ledger) pr = context.get("pr") or {} pr_number = int(pr["number"]) @@ -76,9 +55,7 @@ def update_ledger( entry_recorded_at = str(existing.get("recorded_at") or recorded_at) by_pr[pr_number] = build_entry(context, result, entry_recorded_at) - ledger["version"] = 1 - ledger["entries"] = sorted(by_pr.values(), key=lambda item: (item.get("merged_at") or "", int(item.get("pr") or 0))) - return ledger + return set_sorted_entries(ledger, list(by_pr.values())) def main() -> int: @@ -93,7 +70,7 @@ def main() -> int: ledger_path = Path(args.ledger or context.get("ledger_path") or DEFAULT_LEDGER_PATH) ledger_path.parent.mkdir(parents=True, exist_ok=True) ledger = update_ledger(load_ledger(ledger_path), context, result, now_iso()) - ledger_path.write_text(json.dumps(ledger, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + write_ledger(ledger_path, ledger) return 0 diff --git a/.github/scripts/validate_implementation_output.py b/.github/scripts/validate_implementation_output.py index 55a0a4c..e4055c3 100644 --- a/.github/scripts/validate_implementation_output.py +++ b/.github/scripts/validate_implementation_output.py @@ -4,27 +4,25 @@ from __future__ import annotations import argparse -import json -import re +import sys from pathlib import Path from typing import Any -from implementation_file_filters import TEMP_WORKFLOW_PATHS, is_generated_path +SCRIPT_DIR = Path(__file__).resolve().parent +sys.path.insert(0, str(SCRIPT_DIR)) +from pr_metadata_contracts import ( # noqa: E402 + BASE_METADATA_FIELDS, + load_json_object, + validate_base_metadata, + validate_intended_files, +) -REQUIRED_METADATA_FIELDS = {"branch_name", "pr_title", "pr_summary", "intended_files"} -STRING_METADATA_FIELDS = {"branch_name", "pr_title", "pr_summary"} -CONVENTIONAL_TITLE_RE = re.compile(r"^(feat|fix|docs|style|refactor|perf|test|build|ci|chore)(\([a-z0-9._-]+\))?: .+") +REQUIRED_METADATA_FIELDS = {*BASE_METADATA_FIELDS, "intended_files"} def load_json(path: Path) -> dict[str, Any]: - try: - value = json.loads(path.read_text(encoding="utf-8")) - except json.JSONDecodeError as exc: - raise SystemExit(f"{path} is invalid JSON: {exc}") from exc - if not isinstance(value, dict): - raise SystemExit(f"{path} must contain a JSON object") - return value + return load_json_object(path) def validate_metadata(metadata_path: Path, context_path: Path) -> dict[str, str]: @@ -32,24 +30,8 @@ def validate_metadata(metadata_path: Path, context_path: Path) -> dict[str, str] raise SystemExit("pr-metadata.json was not created") context = load_json(context_path) metadata = load_json(metadata_path) - missing = sorted(REQUIRED_METADATA_FIELDS - set(metadata)) - if missing: - raise SystemExit(f"pr-metadata.json is missing fields: {', '.join(missing)}") - for field in STRING_METADATA_FIELDS: - if not isinstance(metadata.get(field), str) or not metadata[field].strip(): - raise SystemExit(f"pr-metadata.json field {field} must be a non-empty string") - intended_files = metadata.get("intended_files") - if not isinstance(intended_files, list) or not intended_files: - raise SystemExit("pr-metadata.json field intended_files must be a non-empty list") - for index, path in enumerate(intended_files): - if not isinstance(path, str) or not path.strip(): - raise SystemExit(f"pr-metadata.json intended_files[{index}] must be a non-empty string") - if Path(path).is_absolute() or ".." in Path(path).parts: - raise SystemExit(f"pr-metadata.json intended_files[{index}] must be a repository-relative path") - if path in TEMP_WORKFLOW_PATHS: - raise SystemExit(f"pr-metadata.json intended_files[{index}] must not include workflow handoff files") - if is_generated_path(path): - raise SystemExit(f"pr-metadata.json intended_files[{index}] must not include generated/cache files") + validate_base_metadata(metadata, required_fields=REQUIRED_METADATA_FIELDS) + validate_intended_files(metadata) branch_name = metadata["branch_name"].strip() target_branch = str(context.get("target_branch") or "").strip() @@ -66,17 +48,11 @@ def validate_metadata(metadata_path: Path, context_path: Path) -> dict[str, str] f"{branch_prefix}-" ) - if not CONVENTIONAL_TITLE_RE.match(metadata["pr_title"]): - raise SystemExit("pr-metadata.json pr_title must use conventional commit style") - issue_number = context.get("issue_number") expected_first_line = f"Closes #{issue_number}" first_line = metadata["pr_summary"].splitlines()[0] if metadata["pr_summary"].splitlines() else "" if first_line != expected_first_line: raise SystemExit(f"pr-metadata.json pr_summary first line must be exactly {expected_first_line!r}") - if "\n" not in metadata["pr_summary"]: - raise SystemExit("pr-metadata.json pr_summary must be a complete markdown body, not a one-line note") - return {field: metadata[field] for field in REQUIRED_METADATA_FIELDS} diff --git a/.github/scripts/validate_pr_comment_result.py b/.github/scripts/validate_pr_comment_result.py index 80ccd22..724bccf 100644 --- a/.github/scripts/validate_pr_comment_result.py +++ b/.github/scripts/validate_pr_comment_result.py @@ -4,7 +4,6 @@ from __future__ import annotations import argparse -import json import re import sys from pathlib import Path @@ -14,62 +13,30 @@ sys.path.insert(0, str(SCRIPT_DIR)) from commit_implementation_branch import implementation_paths, status_paths # noqa: E402 -from implementation_file_filters import TEMP_WORKFLOW_PATHS, is_generated_path # noqa: E402 +from pr_metadata_contracts import ( # noqa: E402 + BASE_METADATA_FIELDS, + load_json_object, + validate_base_metadata, + validate_intended_files, +) -REQUIRED_METADATA_FIELDS = {"branch_name", "pr_title", "pr_summary", "intended_files"} -STRING_METADATA_FIELDS = {"branch_name", "pr_title", "pr_summary"} -CONVENTIONAL_TITLE_RE = re.compile(r"^(feat|fix|docs|style|refactor|perf|test|build|ci|chore)(\([a-z0-9._-]+\))?: .+") +REQUIRED_METADATA_FIELDS = {*BASE_METADATA_FIELDS, "intended_files"} SENTENCE_END_RE = re.compile(r"[.!?](?:\s+|$)") def load_json(path: Path, *, required: bool = True) -> dict[str, Any]: - if not path.exists(): - if required: - raise SystemExit(f"{path} was not created") - return {} - try: - value = json.loads(path.read_text(encoding="utf-8")) - except json.JSONDecodeError as exc: - raise SystemExit(f"{path} is invalid JSON: {exc}") from exc - if not isinstance(value, dict): - raise SystemExit(f"{path} must contain a JSON object") - return value - - -def validate_intended_path(path: object, index: int) -> str: - if not isinstance(path, str) or not path.strip(): - raise SystemExit(f"pr-metadata.json intended_files[{index}] must be a non-empty string") - normalized = path.strip() - if Path(normalized).is_absolute() or ".." in Path(normalized).parts: - raise SystemExit(f"pr-metadata.json intended_files[{index}] must be a repository-relative path") - if normalized in TEMP_WORKFLOW_PATHS: - raise SystemExit(f"pr-metadata.json intended_files[{index}] must not include workflow handoff files") - if is_generated_path(normalized): - raise SystemExit(f"pr-metadata.json intended_files[{index}] must not include generated/cache files") - return normalized + return load_json_object(path, required=required) def validate_metadata(metadata_path: Path, context_path: Path, candidate_paths: list[str] | None = None) -> dict[str, Any]: context = load_json(context_path) metadata = load_json(metadata_path) - missing = sorted(REQUIRED_METADATA_FIELDS - set(metadata)) - if missing: - raise SystemExit(f"pr-metadata.json is missing fields: {', '.join(missing)}") - for field in STRING_METADATA_FIELDS: - if not isinstance(metadata.get(field), str) or not metadata[field].strip(): - raise SystemExit(f"pr-metadata.json field {field} must be a non-empty string") + validate_base_metadata(metadata, required_fields=REQUIRED_METADATA_FIELDS) if metadata["branch_name"].strip() != str(context.get("agent_push_branch") or "").strip(): raise SystemExit("pr-metadata.json branch_name must equal pr_comment_context.json agent_push_branch") - if not CONVENTIONAL_TITLE_RE.match(metadata["pr_title"]): - raise SystemExit("pr-metadata.json pr_title must use conventional commit style") - if "\n" not in metadata["pr_summary"]: - raise SystemExit("pr-metadata.json pr_summary must be a complete markdown body, not a one-line note") - - raw_files = metadata.get("intended_files") - if not isinstance(raw_files, list) or not raw_files: - raise SystemExit("pr-metadata.json field intended_files must be a non-empty list") - intended = sorted(dict.fromkeys(validate_intended_path(path, index) for index, path in enumerate(raw_files))) + + intended = sorted(dict.fromkeys(validate_intended_files(metadata))) candidates = implementation_paths(candidate_paths if candidate_paths is not None else status_paths()) missing_changes = sorted(set(intended) - set(candidates)) diff --git a/.github/scripts/validate_product_docs_sync_result.py b/.github/scripts/validate_product_docs_sync_result.py index 12a9f18..b0c6595 100644 --- a/.github/scripts/validate_product_docs_sync_result.py +++ b/.github/scripts/validate_product_docs_sync_result.py @@ -9,6 +9,8 @@ from pathlib import Path from typing import Any +from artifact_contracts import write_github_output + VALID_DECISIONS = {"required", "uncertain", "not-needed"} LEDGER_FILE = "docs/product/.product-docs-sync-ledger.json" @@ -87,14 +89,6 @@ def validate_write_surface(decision: str, paths: list[str]) -> list[str]: return product_doc_paths -def write_github_output(path: str | None, values: dict[str, str]) -> None: - if not path: - return - with Path(path).open("a", encoding="utf-8") as handle: - for key, value in values.items(): - handle.write(f"{key}={value}\n") - - def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--result", default="product-docs-sync-result.json") diff --git a/.github/scripts/validate_product_wiki_compile.py b/.github/scripts/validate_product_wiki_compile.py index e5b543f..73b30d4 100644 --- a/.github/scripts/validate_product_wiki_compile.py +++ b/.github/scripts/validate_product_wiki_compile.py @@ -9,6 +9,8 @@ import subprocess from pathlib import Path +from artifact_contracts import write_github_output + WIKI_ROOT = "docs/product/wiki/" REQUIRED_FILES = { @@ -247,14 +249,6 @@ def validate_health_contract(root: Path) -> None: raise SystemExit(f"{relative} uses 待确认 or 开放问题 without a dedicated review section") -def write_github_output(path: str | None, values: dict[str, str]) -> None: - if not path: - return - with Path(path).open("a", encoding="utf-8") as handle: - for key, value in values.items(): - handle.write(f"{key}={value}\n") - - def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--root", default=".") diff --git a/.github/scripts/validate_spec_output.py b/.github/scripts/validate_spec_output.py index 787eb9a..bbe29ea 100644 --- a/.github/scripts/validate_spec_output.py +++ b/.github/scripts/validate_spec_output.py @@ -4,45 +4,34 @@ from __future__ import annotations import argparse -import json -import re import subprocess +import sys from pathlib import Path +SCRIPT_DIR = Path(__file__).resolve().parent +sys.path.insert(0, str(SCRIPT_DIR)) -REQUIRED_METADATA_FIELDS = {"branch_name", "pr_title", "pr_summary"} -CONVENTIONAL_TITLE_RE = re.compile(r"^(feat|fix|docs|style|refactor|perf|test|build|ci|chore)(\([a-z0-9._-]+\))?: .+") +from pr_metadata_contracts import BASE_METADATA_FIELDS, load_json_object, validate_base_metadata # noqa: E402 + + +REQUIRED_METADATA_FIELDS = BASE_METADATA_FIELDS def load_context(path: Path) -> dict[str, object]: - return json.loads(path.read_text(encoding="utf-8")) + return load_json_object(path) def validate_metadata(path: Path, branch_name: str, issue_number: int | None = None) -> dict[str, str]: if not path.exists(): raise SystemExit("pr-metadata.json was not created") - try: - metadata = json.loads(path.read_text(encoding="utf-8")) - except json.JSONDecodeError as exc: - raise SystemExit(f"pr-metadata.json is invalid JSON: {exc}") from exc - if not isinstance(metadata, dict): - raise SystemExit("pr-metadata.json must contain a JSON object") - missing = sorted(REQUIRED_METADATA_FIELDS - set(metadata)) - if missing: - raise SystemExit(f"pr-metadata.json is missing fields: {', '.join(missing)}") - for field in REQUIRED_METADATA_FIELDS: - if not isinstance(metadata.get(field), str) or not metadata[field].strip(): - raise SystemExit(f"pr-metadata.json field {field} must be a non-empty string") + metadata = load_json_object(path, display_name="pr-metadata.json") + validate_base_metadata(metadata, required_fields=REQUIRED_METADATA_FIELDS) if metadata["branch_name"] != branch_name: raise SystemExit( f"pr-metadata.json branch_name must be {branch_name!r}, got {metadata['branch_name']!r}" ) - if not CONVENTIONAL_TITLE_RE.match(metadata["pr_title"]): - raise SystemExit("pr-metadata.json pr_title must use conventional commit style") if issue_number is not None and f"Refs #{issue_number}" not in metadata["pr_summary"]: raise SystemExit(f"pr-metadata.json pr_summary must include Refs #{issue_number}") - if "\n" not in metadata["pr_summary"]: - raise SystemExit("pr-metadata.json pr_summary must be a complete markdown body, not a one-line note") return metadata diff --git a/.github/scripts/write_spec_context.py b/.github/scripts/write_spec_context.py index 8a2e987..0f6cdc9 100644 --- a/.github/scripts/write_spec_context.py +++ b/.github/scripts/write_spec_context.py @@ -12,53 +12,27 @@ 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 +from issue_refs import issue_number_from_text as shared_issue_number_from_text +from issue_refs import resolve_issue_number_from_pr + APPROVED_LABEL = "plan-approved" SPEC_CONTEXT_NONE = "Spec Context: No approved or repository spec context was found for this PR." DIFF_FILE_RE = re.compile(r"^FILE\s+(.+?)\s*$") -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_event(path: str) -> dict[str, Any]: return json.loads(Path(path).read_text(encoding="utf-8")) def issue_number_from_text(text: str) -> int | None: - patterns = [ - r"(? int | None: - for text in ( - pr.get("body") or "", - pr.get("title") or "", - (pr.get("head") or {}).get("ref") or "", - ): - issue_number = issue_number_from_text(text) - if issue_number is not None: - return issue_number - return None + return resolve_issue_number_from_pr(pr) def spec_file_paths(issue_number: int) -> list[str]: