410 lines
15 KiB
Python
410 lines
15 KiB
Python
"""Deterministic Gitea issue-to-release execution engine."""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import signal
|
|
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
|
|
timed_out: bool = False
|
|
|
|
|
|
@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"
|
|
agent_timeout_seconds: float = 1800.0
|
|
test_timeout_seconds: float = 900.0
|
|
termination_grace_seconds: float = 5.0
|
|
|
|
|
|
@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: list[str],
|
|
cwd: Path,
|
|
env: dict[str, str] | None = None,
|
|
timeout_seconds: float | None = None,
|
|
termination_grace_seconds: float = 5.0,
|
|
) -> 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'}"
|
|
|
|
|
|
def parse_command(command: str, name: str) -> list[str]:
|
|
try:
|
|
arguments = shlex.split(command)
|
|
except ValueError as exc:
|
|
raise ValueError(f"invalid {name}: {exc}") from exc
|
|
if not arguments:
|
|
raise ValueError(f"invalid {name}: command is empty")
|
|
return arguments
|
|
|
|
|
|
class ShellRunner:
|
|
def run(
|
|
self,
|
|
command: list[str],
|
|
cwd: Path,
|
|
env: dict[str, str] | None = None,
|
|
timeout_seconds: float | None = None,
|
|
termination_grace_seconds: float = 5.0,
|
|
) -> CommandResult:
|
|
merged_env = os.environ.copy()
|
|
if env:
|
|
merged_env.update(env)
|
|
process = subprocess.Popen(
|
|
command,
|
|
cwd=str(cwd),
|
|
env=merged_env,
|
|
shell=False,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
start_new_session=True,
|
|
)
|
|
try:
|
|
stdout, stderr = process.communicate(timeout=timeout_seconds)
|
|
return CommandResult(process.returncode, stdout, stderr)
|
|
except subprocess.TimeoutExpired:
|
|
os.killpg(process.pid, signal.SIGTERM)
|
|
try:
|
|
stdout, stderr = process.communicate(timeout=termination_grace_seconds)
|
|
except subprocess.TimeoutExpired:
|
|
os.killpg(process.pid, signal.SIGKILL)
|
|
stdout, stderr = process.communicate()
|
|
return CommandResult(process.returncode, stdout, stderr, timed_out=True)
|
|
|
|
|
|
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")
|
|
if exc.code == 401:
|
|
detail += " (authentication rejected; refresh the cron GITEA_TOKEN)"
|
|
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]:
|
|
deadlines = {
|
|
"agent timeout seconds": self.config.agent_timeout_seconds,
|
|
"test timeout seconds": self.config.test_timeout_seconds,
|
|
"termination grace seconds": self.config.termination_grace_seconds,
|
|
}
|
|
for name, value in deadlines.items():
|
|
if value <= 0:
|
|
raise ValueError(f"{name} must be positive")
|
|
|
|
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,
|
|
}
|
|
|
|
agent_command = parse_command(self.config.agent_command, "agent command")
|
|
test_command = parse_command(self.config.test_command, "test command")
|
|
|
|
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(["git", "checkout", "-B", branch], self.config.repo_path)
|
|
if branch_result.returncode != 0:
|
|
raise RuntimeError(f"branch creation failed: {branch_result.stderr.strip()}")
|
|
|
|
agent_result = self.runner.run(
|
|
agent_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,
|
|
},
|
|
timeout_seconds=self.config.agent_timeout_seconds,
|
|
termination_grace_seconds=self.config.termination_grace_seconds,
|
|
)
|
|
if agent_result.timed_out:
|
|
evidence = (agent_result.stdout + "\n" + agent_result.stderr).strip()[-2000:]
|
|
self.store.save(RunState(issue.number, branch, "agent_timed_out", evidence))
|
|
raise RuntimeError(f"coding agent timed out: {evidence}")
|
|
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(
|
|
test_command,
|
|
self.config.repo_path,
|
|
timeout_seconds=self.config.test_timeout_seconds,
|
|
termination_grace_seconds=self.config.termination_grace_seconds,
|
|
)
|
|
evidence = (test_result.stdout + "\n" + test_result.stderr).strip()[-4000:]
|
|
if test_result.timed_out:
|
|
self.store.save(RunState(issue.number, branch, "tests_timed_out", evidence))
|
|
raise RuntimeError(f"tests timed out: {evidence}")
|
|
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(["git", "push", "-u", "origin", 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"))
|
|
parser.add_argument("--agent-timeout", type=float, default=float(os.getenv("RELEASE_AGENT_TIMEOUT", "1800")))
|
|
parser.add_argument("--test-timeout", type=float, default=float(os.getenv("RELEASE_TEST_TIMEOUT", "900")))
|
|
parser.add_argument("--termination-grace", type=float, default=float(os.getenv("RELEASE_TERMINATION_GRACE", "5")))
|
|
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,
|
|
agent_timeout_seconds=args.agent_timeout,
|
|
test_timeout_seconds=args.test_timeout,
|
|
termination_grace_seconds=args.termination_grace,
|
|
)
|
|
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())
|