feat: autonomous issue-to-release execution engine

src/release_engine.py

@@ -0,0 +1,338 @@
+"""Deterministic Gitea issue-to-release execution engine."""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import re
+import shlex
+import subprocess
+from dataclasses import asdict, dataclass
+from pathlib import Path
+from typing import Any, Iterable, Protocol
+from urllib import error, request
+
+
+PRIORITY = {"P0": 0, "P1": 1, "P2": 2}
+
+
+@dataclass(frozen=True)
+class Issue:
+    number: int
+    title: str
+    body: str
+    state: str
+    labels: list[str]
+    assignee: str | None
+
+
+@dataclass(frozen=True)
+class CommandResult:
+    returncode: int
+    stdout: str
+    stderr: str
+
+
+@dataclass(frozen=True)
+class EngineConfig:
+    repo: str
+    agent: str
+    repo_path: Path
+    state_path: Path
+    agent_command: str
+    test_command: str
+    base_branch: str = "main"
+
+
+@dataclass(frozen=True)
+class RunState:
+    issue: int
+    branch: str
+    status: str
+    evidence: str = ""
+    pr_number: int | None = None
+    pr_url: str = ""
+
+
+class Client(Protocol):
+    def list_open_issues(self, repo: str) -> list[Issue]: ...
+    def claim_issue(self, repo: str, number: int, assignee: str) -> None: ...
+    def get_issue(self, repo: str, number: int) -> Issue: ...
+    def find_open_pr(self, repo: str, head: str) -> dict[str, Any] | None: ...
+    def create_pr(self, repo: str, head: str, base: str, title: str, body: str) -> dict[str, Any]: ...
+
+
+class Runner(Protocol):
+    def run(self, command: str, cwd: Path, env: dict[str, str] | None = None) -> CommandResult: ...
+
+
+class StateStore:
+    def __init__(self, path: Path):
+        self.path = Path(path)
+
+    def load(self) -> RunState | None:
+        if not self.path.exists():
+            return None
+        return RunState(**json.loads(self.path.read_text()))
+
+    def save(self, state: RunState) -> None:
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        tmp = self.path.with_name(self.path.name + ".tmp")
+        tmp.write_text(json.dumps(asdict(state), indent=2) + "\n")
+        tmp.replace(self.path)
+
+
+def _label_names(labels: Iterable[Any]) -> list[str]:
+    return [label.get("name", "") if isinstance(label, dict) else str(label) for label in labels]
+
+
+def _priority(issue: Issue) -> tuple[int, int]:
+    rank = min((PRIORITY.get(name, 3) for name in issue.labels), default=3)
+    return rank, issue.number
+
+
+def select_issue(issues: Iterable[Issue], agent: str, issue_number: int | None = None) -> Issue:
+    open_issues = [i for i in issues if i.state == "open"]
+    if issue_number is not None:
+        match = next((i for i in open_issues if i.number == issue_number), None)
+        if match is None:
+            raise ValueError(f"open issue #{issue_number} not found")
+        if match.assignee not in (None, agent):
+            raise ValueError(f"issue #{issue_number} assigned to {match.assignee}")
+        return match
+
+    eligible = [i for i in open_issues if i.assignee in (None, agent)]
+    if not eligible:
+        raise ValueError("no eligible open issues")
+    return sorted(eligible, key=_priority)[0]
+
+
+def branch_name(agent: str, issue: Issue) -> str:
+    slug = re.sub(r"[^a-z0-9]+", "-", issue.title.lower()).strip("-")[:48]
+    actor = re.sub(r"[^a-z0-9]+", "-", agent.lower()).strip("-")
+    return f"{actor}/{issue.number}-{slug or 'issue'}"
+
+
+class ShellRunner:
+    def run(self, command: str, cwd: Path, env: dict[str, str] | None = None) -> CommandResult:
+        merged_env = os.environ.copy()
+        if env:
+            merged_env.update(env)
+        completed = subprocess.run(
+            command,
+            cwd=str(cwd),
+            env=merged_env,
+            shell=True,
+            text=True,
+            capture_output=True,
+        )
+        return CommandResult(completed.returncode, completed.stdout, completed.stderr)
+
+
+class GiteaClient:
+    def __init__(self, api_url: str, token: str):
+        self.api_url = api_url.rstrip("/")
+        self.token = token
+
+    def _request(self, method: str, path: str, body: dict[str, Any] | None = None) -> Any:
+        data = json.dumps(body).encode() if body is not None else None
+        req = request.Request(f"{self.api_url}/{path.lstrip('/')}", data=data, method=method)
+        req.add_header("Accept", "application/json")
+        req.add_header("Authorization", f"token {self.token}")
+        if data is not None:
+            req.add_header("Content-Type", "application/json")
+        try:
+            with request.urlopen(req, timeout=20) as response:
+                raw = response.read()
+                return json.loads(raw) if raw else None
+        except error.HTTPError as exc:
+            detail = exc.read().decode(errors="replace")
+            raise RuntimeError(f"Gitea {method} {path} failed: HTTP {exc.code}: {detail}") from exc
+
+    @staticmethod
+    def _issue(item: dict[str, Any]) -> Issue:
+        assignee = (item.get("assignee") or {}).get("login")
+        return Issue(
+            number=item["number"],
+            title=item["title"],
+            body=item.get("body", ""),
+            state=item["state"],
+            labels=_label_names(item.get("labels", [])),
+            assignee=assignee,
+        )
+
+    def list_open_issues(self, repo: str) -> list[Issue]:
+        items = self._request("GET", f"repos/{repo}/issues?state=open&limit=100")
+        return [self._issue(item) for item in items if not item.get("pull_request")]
+
+    def claim_issue(self, repo: str, number: int, assignee: str) -> None:
+        self._request("PATCH", f"repos/{repo}/issues/{number}", {"assignee": assignee})
+
+    def get_issue(self, repo: str, number: int) -> Issue:
+        return self._issue(self._request("GET", f"repos/{repo}/issues/{number}"))
+
+    def find_open_pr(self, repo: str, head: str) -> dict[str, Any] | None:
+        pulls = self._request("GET", f"repos/{repo}/pulls?state=open&limit=100")
+        for pull in pulls:
+            if pull.get("head", {}).get("label") == head or pull.get("head", {}).get("ref") == head:
+                return pull
+        return None
+
+    def create_pr(self, repo: str, head: str, base: str, title: str, body: str) -> dict[str, Any]:
+        return self._request("POST", f"repos/{repo}/pulls", {
+            "head": head,
+            "base": base,
+            "title": title,
+            "body": body,
+        })
+
+
+class ReleaseEngine:
+    def __init__(self, config: EngineConfig, client: Client, runner: Runner):
+        self.config = config
+        self.client = client
+        self.runner = runner
+        self.store = StateStore(config.state_path)
+
+    def execute(self, issue_number: int | None = None, dry_run: bool = False) -> dict[str, Any]:
+        issue = select_issue(self.client.list_open_issues(self.config.repo), self.config.agent, issue_number)
+        branch = branch_name(self.config.agent, issue)
+
+        existing_pr = self.client.find_open_pr(self.config.repo, branch)
+        if existing_pr:
+            return {
+                "issue": issue.number,
+                "branch": branch,
+                "pr_number": existing_pr["number"],
+                "pr_url": existing_pr.get("html_url", ""),
+                "reused": True,
+            }
+
+        if dry_run:
+            return {
+                "dry_run": True,
+                "issue": issue.number,
+                "title": issue.title,
+                "branch": branch,
+                "repo": self.config.repo,
+                "agent": self.config.agent,
+            }
+
+        self.client.claim_issue(self.config.repo, issue.number, self.config.agent)
+        verified = self.client.get_issue(self.config.repo, issue.number)
+        if verified.assignee != self.config.agent:
+            raise RuntimeError(
+                f"claim verification failed: expected {self.config.agent}, got {verified.assignee}"
+            )
+        self.store.save(RunState(issue.number, branch, "claimed"))
+
+        branch_result = self.runner.run(f"git checkout -B {shlex.quote(branch)}", self.config.repo_path)
+        if branch_result.returncode != 0:
+            raise RuntimeError(f"branch creation failed: {branch_result.stderr.strip()}")
+
+        context = {
+            "issue_number": issue.number,
+            "issue_title": issue.title,
+            "issue_body": issue.body,
+            "repo": self.config.repo,
+            "branch": branch,
+        }
+        command = self.config.agent_command.format(**context)
+        agent_result = self.runner.run(command, self.config.repo_path, {
+            "RELEASE_ISSUE_NUMBER": str(issue.number),
+            "RELEASE_ISSUE_TITLE": issue.title,
+            "RELEASE_ISSUE_BODY": issue.body,
+            "RELEASE_REPO": self.config.repo,
+            "RELEASE_BRANCH": branch,
+        })
+        if agent_result.returncode != 0:
+            self.store.save(RunState(issue.number, branch, "agent_failed", agent_result.stderr[-2000:]))
+            raise RuntimeError(f"coding agent failed: {agent_result.stderr.strip()}")
+        self.store.save(RunState(issue.number, branch, "agent_complete"))
+
+        test_result = self.runner.run(self.config.test_command, self.config.repo_path)
+        evidence = (test_result.stdout + "\n" + test_result.stderr).strip()[-4000:]
+        if test_result.returncode != 0:
+            self.store.save(RunState(issue.number, branch, "tests_failed", evidence))
+            raise RuntimeError(f"tests failed: {evidence}")
+        self.store.save(RunState(issue.number, branch, "tests_passed", evidence))
+
+        push_result = self.runner.run(f"git push -u origin {shlex.quote(branch)}", self.config.repo_path)
+        if push_result.returncode != 0:
+            self.store.save(RunState(issue.number, branch, "push_failed", push_result.stderr[-2000:]))
+            raise RuntimeError(f"push failed: {push_result.stderr.strip()}")
+        self.store.save(RunState(issue.number, branch, "pushed", evidence))
+
+        pr_body = (
+            f"Closes #{issue.number}\n\n"
+            "## Automated validation\n"
+            f"- Command: `{self.config.test_command}`\n"
+            f"- Result: PASS\n\n"
+            "```text\n"
+            f"{evidence[-2000:]}\n"
+            "```"
+        )
+        pr = self.client.create_pr(
+            self.config.repo,
+            branch,
+            self.config.base_branch,
+            f"feat: {issue.title}",
+            pr_body,
+        )
+        final = RunState(
+            issue.number,
+            branch,
+            "pr_opened",
+            evidence,
+            pr_number=pr["number"],
+            pr_url=pr.get("html_url", ""),
+        )
+        self.store.save(final)
+        return {
+            "issue": issue.number,
+            "branch": branch,
+            "pr_number": pr["number"],
+            "pr_url": pr.get("html_url", ""),
+            "tests": "passed",
+        }
+
+
+def _parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--repo", required=True)
+    parser.add_argument("--agent", default=os.getenv("RELEASE_AGENT", "timmy"))
+    parser.add_argument("--repo-path", type=Path, default=Path.cwd())
+    parser.add_argument("--state", type=Path, default=Path(".release-engine/state.json"))
+    parser.add_argument("--issue", type=int)
+    parser.add_argument("--dry-run", action="store_true")
+    parser.add_argument("--api", default=os.getenv("GITEA_API", "http://127.0.0.1:3000/api/v1"))
+    parser.add_argument("--agent-command", default=os.getenv("RELEASE_AGENT_COMMAND", ""))
+    parser.add_argument("--test-command", default=os.getenv("RELEASE_TEST_COMMAND", "python3 -m pytest tests/ -q"))
+    return parser.parse_args()
+
+
+def main() -> int:
+    args = _parse_args()
+    token = os.getenv("GITEA_TOKEN", "")
+    if not token:
+        raise SystemExit("GITEA_TOKEN is required")
+    if not args.dry_run and not args.agent_command:
+        raise SystemExit("RELEASE_AGENT_COMMAND is required outside --dry-run")
+    config = EngineConfig(
+        repo=args.repo,
+        agent=args.agent,
+        repo_path=args.repo_path.resolve(),
+        state_path=args.state,
+        agent_command=args.agent_command,
+        test_command=args.test_command,
+    )
+    result = ReleaseEngine(config, GiteaClient(args.api, token), ShellRunner()).execute(
+        issue_number=args.issue,
+        dry_run=args.dry_run,
+    )
+    print(json.dumps(result, indent=2))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())