from __future__ import annotations import argparse import fnmatch import json import os import re import shlex import subprocess import sys from dataclasses import dataclass from pathlib import Path from typing import Any, Iterable, Sequence try: import tomllib except ModuleNotFoundError: # pragma: no cover - Python < 3.11 is unsupported tomllib = None STORY_RE = re.compile(r"^(US(?P\d{2})-(?P\d{2}))\s+—\s+(?P.+)$") SECRET_PATTERNS = ( re.compile(r"(?i)(api[_-]?key|access[_-]?token|client[_-]?secret|password)\s*[:=]\s*['\"]?[A-Za-z0-9_./+\-=]{12,}"), re.compile(r"-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----"), re.compile(r"(?i)authorization:\s*(?:bearer|token)\s+[A-Za-z0-9._~+\-/=]{12,}"), ) class WorkItemError(RuntimeError): pass class Runner: def run( self, args: Sequence[str], *, cwd: Path, check: bool = True, input_text: str | None = None, ) -> subprocess.CompletedProcess[str]: result = subprocess.run( list(args), cwd=cwd, text=True, input=input_text, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) if check and result.returncode: command = " ".join(shlex.quote(part) for part in args) detail = (result.stderr or result.stdout).strip() raise WorkItemError(f"Command failed ({result.returncode}): {command}\n{detail}") return result def run_shell(self, command: str, *, cwd: Path) -> subprocess.CompletedProcess[str]: result = subprocess.run(command, cwd=cwd, text=True, shell=True) if result.returncode: raise WorkItemError(f"Test command failed ({result.returncode}): {command}") return result @dataclass(frozen=True) class Config: repo_slug: str login: str assignee: str remote: str = "origin" main_branch: str = "main" branch_prefix: str = "us" require_ci: bool = False max_file_bytes: int = 5_000_000 required_tests: tuple[str, ...] = () denied_patterns: tuple[str, ...] = () allowed_patterns: tuple[str, ...] = () @classmethod def load(cls, path: Path) -> "Config": if tomllib is None: raise WorkItemError("Python 3.11 or newer is required") if not path.is_file(): raise WorkItemError(f"Missing workflow configuration: {path}") with path.open("rb") as handle: raw = tomllib.load(handle) repo = raw.get("repository", {}) workflow = raw.get("workflow", {}) safety = raw.get("safety", {}) required = ("slug", "login", "assignee") missing = [key for key in required if not repo.get(key)] if missing: raise WorkItemError(f"Missing repository configuration: {', '.join(missing)}") return cls( repo_slug=repo["slug"], login=repo["login"], assignee=repo["assignee"], remote=repo.get("remote", "origin"), main_branch=repo.get("main_branch", "main"), branch_prefix=workflow.get("branch_prefix", "us"), require_ci=bool(workflow.get("require_ci", False)), max_file_bytes=int(safety.get("max_file_bytes", 5_000_000)), required_tests=tuple(workflow.get("required_tests", ())), denied_patterns=tuple(safety.get("deny", ())), allowed_patterns=tuple(safety.get("allow", ())), ) @dataclass(frozen=True) class Story: number: int story_id: str title: str epic: int sequence: int labels: frozenset[str] assignees: tuple[str, ...] raw: dict[str, Any] @classmethod def from_issue(cls, issue: dict[str, Any]) -> "Story | None": match = STORY_RE.match(issue.get("title", "")) if not match: return None labels = frozenset(label["name"] for label in issue.get("labels") or ()) assignees = tuple(user["login"] for user in issue.get("assignees") or ()) return cls( number=int(issue["number"]), story_id=match.group(1), title=match.group("title"), epic=int(match.group("epic")), sequence=int(match.group("story")), labels=labels, assignees=assignees, raw=issue, ) @property def branch_slug(self) -> str: value = re.sub(r"[^a-z0-9]+", "-", self.title.lower()).strip("-") return value[:48].rstrip("-") class GitRepo: def __init__(self, root: Path, config: Config, runner: Runner) -> None: self.root = root self.config = config self.runner = runner def git(self, *args: str, check: bool = True) -> str: return self.runner.run(("git", *args), cwd=self.root, check=check).stdout.strip() def verify(self) -> None: actual = Path(self.git("rev-parse", "--show-toplevel")).resolve() if actual != self.root.resolve(): raise WorkItemError(f"Run from repository root {actual}; current root is {self.root}") remote_url = self.git("remote", "get-url", self.config.remote) normalized = remote_url.removesuffix(".git").replace(":", "/") if self.config.repo_slug not in normalized: raise WorkItemError( f"Remote {self.config.remote!r} does not match {self.config.repo_slug!r}: {remote_url}" ) def ensure_clean(self) -> None: if self.git("status", "--porcelain=v1"): raise WorkItemError("Working tree must be clean before claiming a story") def update_main(self) -> None: self.git("fetch", "--prune", self.config.remote) self.git("switch", self.config.main_branch) self.git("pull", "--ff-only", self.config.remote, self.config.main_branch) def current_branch(self) -> str: return self.git("branch", "--show-current") def create_story_branch(self, story: Story) -> str: branch = f"{self.config.branch_prefix}/{story.story_id}-{story.branch_slug}" self.git("switch", "-c", branch) return branch def changed_paths(self) -> list[Path]: output = self.git("status", "--porcelain=v1", "--untracked-files=all", "-z") if not output: return [] entries = output.split("\0") paths: list[Path] = [] skip_rename_source = False for entry in entries: if not entry: continue if skip_rename_source: skip_rename_source = False continue if len(entry) < 4: continue status, name = entry[:2], entry[3:] paths.append(Path(name)) if "R" in status or "C" in status: skip_rename_source = True return paths def assert_safe_changes(self) -> list[Path]: paths = self.changed_paths() if not paths: raise WorkItemError("No changes to submit") violations: list[str] = [] for relative in paths: portable = relative.as_posix() allowed = any(fnmatch.fnmatch(portable, pattern) for pattern in self.config.allowed_patterns) denied = any(fnmatch.fnmatch(portable, pattern) for pattern in self.config.denied_patterns) if denied and not allowed: violations.append(f"denied path: {portable}") continue candidate = self.root / relative if candidate.is_file() and candidate.stat().st_size > self.config.max_file_bytes: violations.append(f"file exceeds {self.config.max_file_bytes} bytes: {portable}") continue if candidate.is_file() and candidate.stat().st_size <= self.config.max_file_bytes: try: content = candidate.read_text(encoding="utf-8") except (UnicodeDecodeError, OSError): content = "" for pattern in SECRET_PATTERNS: if pattern.search(content): violations.append(f"possible secret in: {portable}") break if violations: raise WorkItemError("Unsafe changes detected:\n- " + "\n- ".join(violations)) self.git("diff", "--check") return paths def stage_commit_push(self, story: Story, branch: str) -> str: self.git("add", "--all") staged = self.git("diff", "--cached", "--name-only") if not staged: raise WorkItemError("Nothing was staged") subject = f"{story.story_id}: {story.title}" self.git("commit", "-m", subject) self.git("push", "--set-upstream", self.config.remote, branch) return self.git("rev-parse", "HEAD") class Gitea: def __init__(self, root: Path, config: Config, runner: Runner) -> None: self.root = root self.config = config self.runner = runner def tea(self, *args: str) -> str: command = ("tea", *args, "--login", self.config.login) return self.runner.run(command, cwd=self.root).stdout.strip() def api(self, endpoint: str, *, method: str = "GET", data: Any = None) -> Any: args = ["api", endpoint, "--method", method] if data is not None: args.extend(("--data", json.dumps(data))) output = self.tea(*args) return json.loads(output) if output else None @property def base(self) -> str: return f"/repos/{self.config.repo_slug}" def open_stories(self) -> list[Story]: issues = self.api(f"{self.base}/issues?state=open&type=issues&limit=100") stories = [story for issue in issues if (story := Story.from_issue(issue))] return sorted(stories, key=lambda story: (story.epic, story.sequence, story.number)) def dependencies_closed(self, issue_number: int) -> bool: dependencies = self.api(f"{self.base}/issues/{issue_number}/dependencies?limit=100") return all(issue.get("state") == "closed" for issue in dependencies) def next_story(self) -> Story: excluded = {"status/in-progress", "status/review", "status/blocked", "status/done"} for story in self.open_stories(): if story.labels & excluded or story.assignees: continue if self.dependencies_closed(story.number): return story raise WorkItemError("No eligible open user story is available") def edit_labels(self, issue: int, *, add: Iterable[str], remove: Iterable[str]) -> None: args = ["issues", "edit", str(issue), "--repo", self.config.repo_slug] add_value = ",".join(add) remove_value = ",".join(remove) if add_value: args.extend(("--add-labels", add_value)) if remove_value: args.extend(("--remove-labels", remove_value)) self.tea(*args) def claim(self, story: Story) -> None: self.tea( "issues", "edit", str(story.number), "--repo", self.config.repo_slug, "--add-assignees", self.config.assignee, "--add-labels", "status/in-progress", "--remove-labels", "status/backlog,status/ready,status/blocked,status/review,status/done", ) def comment(self, issue: int, body: str) -> None: self.tea("comments", "add", str(issue), body, "--repo", self.config.repo_slug) def create_pr(self, story: Story, branch: str, tests: Sequence[str], commit: str) -> dict[str, Any]: body = "\n".join(( f"Implements #{story.number}", "", "Tests:", *(f"- `{command}`" for command in tests), "", f"Commit: `{commit}`", )) output = self.tea( "pulls", "create", "--repo", self.config.repo_slug, "--head", branch, "--base", self.config.main_branch, "--title", f"{story.story_id}: {story.title}", "--description", body, "--output", "json", ) try: return json.loads(output) except json.JSONDecodeError as error: raise WorkItemError(f"Could not parse created pull request: {output}") from error def find_open_pr(self, branch: str) -> dict[str, Any] | None: pulls = self.api(f"{self.base}/pulls?state=open&limit=50") for pull in pulls: if (pull.get("head") or {}).get("ref") == branch: return pull return None def pr(self, number: int) -> dict[str, Any]: return self.api(f"{self.base}/pulls/{number}") def merge_pr(self, number: int) -> None: self.tea("pulls", "merge", str(number), "--repo", self.config.repo_slug, "--style", "squash") def require_ci_success(self, pr: dict[str, Any]) -> None: if not self.config.require_ci: return sha = pr["head"]["sha"] combined = self.api(f"{self.base}/commits/{sha}/status") if combined.get("state") != "success": raise WorkItemError(f"Required CI is not successful: {combined.get('state', 'unknown')}") def mark_review(self, issue: int, pr: dict[str, Any], tests: Sequence[str]) -> None: self.edit_labels( issue, add=("status/review",), remove=("status/in-progress", "status/backlog", "status/ready", "status/blocked", "status/done"), ) number = pr.get("index") or pr.get("number") url = pr.get("url") or pr.get("html_url") or f"PR #{number}" self.comment(issue, f"Submitted for review: {url}\n\nTests passed:\n" + "\n".join(f"- `{x}`" for x in tests)) def mark_done(self, issue: int, pr_number: int) -> None: self.edit_labels( issue, add=("status/done",), remove=("status/in-progress", "status/backlog", "status/ready", "status/blocked", "status/review"), ) self.comment(issue, f"Completed and merged via PR #{pr_number}.") self.tea("issues", "close", str(issue), "--repo", self.config.repo_slug) class Workflow: STATE_NAME = "work-item-state.json" def __init__(self, root: Path, config: Config, runner: Runner | None = None) -> None: self.root = root.resolve() self.config = config self.runner = runner or Runner() self.git = GitRepo(self.root, config, self.runner) self.gitea = Gitea(self.root, config, self.runner) @property def state_path(self) -> Path: git_dir = self.git.git("rev-parse", "--git-dir") path = Path(git_dir) if not path.is_absolute(): path = self.root / path return path / self.STATE_NAME def load_state(self) -> dict[str, Any]: if not self.state_path.is_file(): raise WorkItemError("No story is currently claimed") return json.loads(self.state_path.read_text(encoding="utf-8")) def save_state(self, state: dict[str, Any]) -> None: temporary = self.state_path.with_suffix(".tmp") temporary.write_text(json.dumps(state, indent=2, sort_keys=True) + "\n", encoding="utf-8") temporary.replace(self.state_path) def clear_state(self) -> None: self.state_path.unlink(missing_ok=True) def next(self) -> Story: self.git.verify() return self.gitea.next_story() def claim(self) -> dict[str, Any]: self.git.verify() self.git.ensure_clean() if self.state_path.exists(): raise WorkItemError("A story is already claimed; complete or block it first") self.git.update_main() story = self.gitea.next_story() branch = self.git.create_story_branch(story) try: self.gitea.claim(story) except Exception: self.git.git("switch", self.config.main_branch) self.git.git("branch", "-D", branch) raise state = { "issue": story.number, "story_id": story.story_id, "title": story.title, "branch": branch, "status": "in-progress", } self.save_state(state) self.gitea.comment(story.number, f"Claimed for implementation on branch `{branch}`.") return state def status(self) -> dict[str, Any]: self.git.verify() state = self.load_state() return {**state, "current_branch": self.git.current_branch(), "changes": [p.as_posix() for p in self.git.changed_paths()]} def submit(self, tests: Sequence[str], *, confirmed: bool) -> dict[str, Any]: self.git.verify() state = self.load_state() if self.git.current_branch() != state["branch"]: raise WorkItemError(f"Expected branch {state['branch']}, found {self.git.current_branch()}") story = Story( number=int(state["issue"]), story_id=state["story_id"], title=state["title"], epic=int(state["story_id"][2:4]), sequence=int(state["story_id"][5:7]), labels=frozenset(), assignees=(), raw={}, ) if state.get("status") == "pushed": commands = tuple(state["tests"]) commit = state["commit"] elif state.get("status") == "in-progress": commands = tuple(dict.fromkeys((*self.config.required_tests, *tests))) if not commands: raise WorkItemError("At least one automated test command is required") self.git.assert_safe_changes() for command in commands: self.runner.run_shell(command, cwd=self.root) paths = self.git.assert_safe_changes() if not confirmed: summary = self.git.git("diff", "--stat") raise WorkItemError( "Review the changes and rerun with --yes to commit:\n" f"{summary}\nFiles: {', '.join(path.as_posix() for path in paths)}" ) commit = self.git.stage_commit_push(story, state["branch"]) state.update({"status": "pushed", "commit": commit, "tests": list(commands)}) self.save_state(state) else: raise WorkItemError(f"Story cannot be submitted from status {state.get('status')!r}") pr = self.gitea.find_open_pr(state["branch"]) if pr is None: pr = self.gitea.create_pr(story, state["branch"], commands, commit) pr_number = int(pr.get("index") or pr.get("number")) state.update({"status": "review", "pr": pr_number, "commit": commit, "tests": list(commands)}) self.save_state(state) self.gitea.mark_review(story.number, pr, commands) return state def complete(self, *, merge: bool) -> dict[str, Any]: self.git.verify() state = self.load_state() if state.get("status") != "review" or not state.get("pr"): raise WorkItemError("The claimed story has not been submitted for review") pr = self.gitea.pr(int(state["pr"])) self.gitea.require_ci_success(pr) if not pr.get("merged"): if not merge: raise WorkItemError("Pull request is not merged; rerun with --merge after review") self.gitea.merge_pr(int(state["pr"])) pr = self.gitea.pr(int(state["pr"])) if not pr.get("merged"): raise WorkItemError("Pull request merge could not be verified") self.gitea.mark_done(int(state["issue"]), int(state["pr"])) branch = state["branch"] self.git.git("switch", self.config.main_branch) self.git.git("pull", "--ff-only", self.config.remote, self.config.main_branch) # A squash-merged PR is verified by Gitea but its feature commit is not an # ancestor of main, so ordinary `-d` would incorrectly reject safe cleanup. self.git.git("branch", "-D", branch) result = {**state, "status": "done"} self.clear_state() return result def block(self, reason: str) -> dict[str, Any]: self.git.verify() if not reason.strip(): raise WorkItemError("A non-empty blocking reason is required") state = self.load_state() self.gitea.edit_labels( int(state["issue"]), add=("status/blocked",), remove=("status/in-progress", "status/ready", "status/review", "status/done"), ) self.gitea.comment(int(state["issue"]), f"Blocked: {reason.strip()}") state.update({"status": "blocked", "reason": reason.strip()}) self.save_state(state) return state def find_root(start: Path) -> Path: result = subprocess.run( ("git", "rev-parse", "--show-toplevel"), cwd=start, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) if result.returncode: raise WorkItemError("This command must run inside the photoanalyzer Git repository") return Path(result.stdout.strip()).resolve() def parser() -> argparse.ArgumentParser: result = argparse.ArgumentParser(prog="work-item", description="Safe issue/Git workflow helper") result.add_argument("--config", default=".work-item.toml") commands = result.add_subparsers(dest="command", required=True) commands.add_parser("next", help="Show the top eligible user story") commands.add_parser("claim", help="Claim the top story and create its branch") commands.add_parser("status", help="Show the current claim and Git state") submit = commands.add_parser("submit", help="Test, commit, push, and open a pull request") submit.add_argument("--test", action="append", default=[], help="Additional automated test command; repeatable") submit.add_argument("--yes", action="store_true", help="Confirm the reviewed diff may be committed") complete = commands.add_parser("complete", help="Verify merge, close the issue, and return to main") complete.add_argument("--merge", action="store_true", help="Merge the reviewed PR before completing") block = commands.add_parser("block", help="Mark the current story blocked") block.add_argument("--reason", required=True) return result def main(argv: Sequence[str] | None = None) -> int: args = parser().parse_args(argv) try: root = find_root(Path.cwd()) config_path = Path(args.config) if not config_path.is_absolute(): config_path = root / config_path workflow = Workflow(root, Config.load(config_path)) if args.command == "next": story = workflow.next() payload: Any = {"issue": story.number, "story_id": story.story_id, "title": story.title} elif args.command == "claim": payload = workflow.claim() elif args.command == "status": payload = workflow.status() elif args.command == "submit": payload = workflow.submit(args.test, confirmed=args.yes) elif args.command == "complete": payload = workflow.complete(merge=args.merge) else: payload = workflow.block(args.reason) print(json.dumps(payload, indent=2, sort_keys=True)) return 0 except WorkItemError as error: print(f"work-item: {error}", file=sys.stderr) return 2 if __name__ == "__main__": raise SystemExit(main())