595 lines
23 KiB
Python
595 lines
23 KiB
Python
"""Deterministic Gitea issue-to-release execution engine."""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import errno
|
|
import fcntl
|
|
import json
|
|
import os
|
|
import re
|
|
import signal
|
|
import shlex
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
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}
|
|
RUNTIME_ENV_NAMES = (
|
|
"HOME", "LANG", "LC_ALL", "LC_CTYPE", "LOGNAME", "SHELL",
|
|
"SSH_AUTH_SOCK", "TMPDIR", "USER",
|
|
)
|
|
|
|
|
|
@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
|
|
agent_env_vars: tuple[str, ...] = ()
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RunState:
|
|
issue: int
|
|
branch: str
|
|
status: str
|
|
evidence: str = ""
|
|
pr_number: int | None = None
|
|
pr_url: str = ""
|
|
schema_version: int = 1
|
|
repo: str = ""
|
|
head_sha: 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)
|
|
temporary_path = None
|
|
try:
|
|
with tempfile.NamedTemporaryFile(
|
|
mode="w",
|
|
encoding="utf-8",
|
|
dir=self.path.parent,
|
|
prefix=self.path.name + ".",
|
|
suffix=".tmp",
|
|
delete=False,
|
|
) as temporary:
|
|
temporary_path = Path(temporary.name)
|
|
temporary.write(json.dumps(asdict(state), indent=2) + "\n")
|
|
temporary.flush()
|
|
os.fsync(temporary.fileno())
|
|
os.replace(temporary_path, self.path)
|
|
directory = os.open(self.path.parent, os.O_RDONLY | os.O_DIRECTORY)
|
|
try:
|
|
os.fsync(directory)
|
|
finally:
|
|
os.close(directory)
|
|
finally:
|
|
if temporary_path is not None:
|
|
temporary_path.unlink(missing_ok=True)
|
|
|
|
|
|
class RepositoryLease:
|
|
"""Non-blocking, process-backed single-writer lease for one worktree."""
|
|
|
|
def __init__(self, repo_path: Path):
|
|
self.path = Path(repo_path).resolve() / ".release-engine" / "run.lock"
|
|
self._file = None
|
|
|
|
def __enter__(self):
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
lock_file = self.path.open("a+", encoding="utf-8")
|
|
try:
|
|
fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
except OSError as exc:
|
|
if exc.errno in (errno.EACCES, errno.EAGAIN):
|
|
detail = ""
|
|
try:
|
|
lock_file.seek(0)
|
|
metadata = json.loads(lock_file.read(512))
|
|
owner_pid = int(metadata["pid"])
|
|
acquired_at = float(metadata["acquired_at"])
|
|
detail = f" (owner pid={owner_pid}, acquired_at={acquired_at:.3f})"
|
|
except (KeyError, TypeError, ValueError, json.JSONDecodeError):
|
|
pass
|
|
lock_file.close()
|
|
raise RuntimeError(f"release run already active{detail}") from exc
|
|
lock_file.close()
|
|
raise
|
|
self._file = lock_file
|
|
lock_file.seek(0)
|
|
lock_file.truncate()
|
|
lock_file.write(json.dumps({"pid": os.getpid(), "acquired_at": time.time()}) + "\n")
|
|
lock_file.flush()
|
|
os.fsync(lock_file.fileno())
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, traceback):
|
|
if self._file is not None:
|
|
fcntl.flock(self._file.fileno(), fcntl.LOCK_UN)
|
|
self._file.close()
|
|
self._file = None
|
|
|
|
|
|
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
|
|
|
|
|
|
def subprocess_environment(
|
|
allowed_names: Iterable[str] = (),
|
|
overrides: dict[str, str] | None = None,
|
|
) -> dict[str, str]:
|
|
"""Build an explicit child environment without inheriting parent secrets."""
|
|
environment = {"PATH": os.environ.get("PATH", os.defpath)}
|
|
for name in (*RUNTIME_ENV_NAMES, *allowed_names):
|
|
if name in os.environ:
|
|
environment[name] = os.environ[name]
|
|
if overrides:
|
|
environment.update(overrides)
|
|
return environment
|
|
|
|
|
|
def validate_agent_env_vars(names: Iterable[str]) -> None:
|
|
for name in names:
|
|
if (
|
|
re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", name) is None
|
|
or name == "GITEA_TOKEN"
|
|
or name.startswith("RELEASE_")
|
|
):
|
|
raise ValueError(f"invalid agent environment variable: {name}")
|
|
|
|
|
|
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:
|
|
process = subprocess.Popen(
|
|
command,
|
|
cwd=str(cwd),
|
|
env=dict(env or {}),
|
|
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]:
|
|
with RepositoryLease(self.config.repo_path):
|
|
return self._execute_locked(issue_number=issue_number, dry_run=dry_run)
|
|
|
|
def _execute_locked(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")
|
|
validate_agent_env_vars(self.config.agent_env_vars)
|
|
|
|
issue = select_issue(self.client.list_open_issues(self.config.repo), self.config.agent, issue_number)
|
|
branch = branch_name(self.config.agent, issue)
|
|
head_sha = ""
|
|
|
|
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")
|
|
runtime_env = subprocess_environment()
|
|
agent_env = subprocess_environment(
|
|
self.config.agent_env_vars,
|
|
{
|
|
"RELEASE_ISSUE_NUMBER": str(issue.number),
|
|
"RELEASE_ISSUE_TITLE": issue.title,
|
|
"RELEASE_ISSUE_BODY": issue.body,
|
|
"RELEASE_REPO": self.config.repo,
|
|
"RELEASE_BRANCH": branch,
|
|
},
|
|
)
|
|
checkpoint = self.store.load()
|
|
resumable_statuses = {"agent_complete", "tests_passed", "pushed"}
|
|
if checkpoint is not None and checkpoint.status in resumable_statuses:
|
|
if (
|
|
checkpoint.schema_version != 1
|
|
or checkpoint.repo != self.config.repo
|
|
or checkpoint.issue != issue.number
|
|
or checkpoint.branch != branch
|
|
or not checkpoint.head_sha
|
|
):
|
|
raise RuntimeError("release checkpoint does not match this run")
|
|
branch_result = self.runner.run(
|
|
["git", "checkout", branch], self.config.repo_path, runtime_env
|
|
)
|
|
if branch_result.returncode != 0:
|
|
raise RuntimeError(f"checkpoint branch checkout failed: {branch_result.stderr.strip()}")
|
|
head_result = self.runner.run(
|
|
["git", "rev-parse", "HEAD"], self.config.repo_path, runtime_env
|
|
)
|
|
if head_result.returncode != 0 or head_result.stdout.strip() != checkpoint.head_sha:
|
|
raise RuntimeError("release checkpoint commit does not match the local branch")
|
|
status_result = self.runner.run(
|
|
["git", "status", "--porcelain"], self.config.repo_path, runtime_env
|
|
)
|
|
if status_result.returncode != 0 or status_result.stdout.strip():
|
|
raise RuntimeError("release checkpoint worktree is not clean")
|
|
head_sha = checkpoint.head_sha
|
|
evidence = checkpoint.evidence
|
|
else:
|
|
checkpoint = None
|
|
evidence = ""
|
|
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, runtime_env
|
|
)
|
|
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,
|
|
agent_env,
|
|
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()}")
|
|
head_result = self.runner.run(
|
|
["git", "rev-parse", "HEAD"], self.config.repo_path, runtime_env
|
|
)
|
|
head_sha = head_result.stdout.strip()
|
|
if head_result.returncode != 0 or not head_sha:
|
|
raise RuntimeError("could not identify the release checkpoint commit")
|
|
self.store.save(RunState(
|
|
issue.number, branch, "agent_complete",
|
|
repo=self.config.repo, head_sha=head_sha,
|
|
))
|
|
|
|
if checkpoint is None or checkpoint.status == "agent_complete":
|
|
test_result = self.runner.run(
|
|
test_command,
|
|
self.config.repo_path,
|
|
runtime_env,
|
|
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,
|
|
repo=self.config.repo, head_sha=head_sha,
|
|
))
|
|
raise RuntimeError(f"tests timed out: {evidence}")
|
|
if test_result.returncode != 0:
|
|
self.store.save(RunState(
|
|
issue.number, branch, "tests_failed", evidence,
|
|
repo=self.config.repo, head_sha=head_sha,
|
|
))
|
|
raise RuntimeError(f"tests failed: {evidence}")
|
|
self.store.save(RunState(
|
|
issue.number, branch, "tests_passed", evidence,
|
|
repo=self.config.repo, head_sha=head_sha,
|
|
))
|
|
|
|
if checkpoint is not None and checkpoint.status == "pushed":
|
|
remote_result = self.runner.run(
|
|
["git", "ls-remote", "--heads", "origin", branch],
|
|
self.config.repo_path,
|
|
runtime_env,
|
|
)
|
|
expected_remote = f"{checkpoint.head_sha}\trefs/heads/{branch}"
|
|
if remote_result.returncode != 0 or remote_result.stdout.strip() != expected_remote:
|
|
raise RuntimeError("release checkpoint commit does not match the remote branch")
|
|
else:
|
|
push_result = self.runner.run(
|
|
["git", "push", "-u", "origin", branch], self.config.repo_path, runtime_env
|
|
)
|
|
if push_result.returncode != 0:
|
|
self.store.save(RunState(
|
|
issue.number, branch, "push_failed", push_result.stderr[-2000:],
|
|
repo=self.config.repo, head_sha=head_sha,
|
|
))
|
|
raise RuntimeError(f"push failed: {push_result.stderr.strip()}")
|
|
self.store.save(RunState(
|
|
issue.number, branch, "pushed", evidence,
|
|
repo=self.config.repo, head_sha=head_sha,
|
|
))
|
|
|
|
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", ""),
|
|
repo=self.config.repo,
|
|
head_sha=head_sha,
|
|
)
|
|
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")))
|
|
parser.add_argument(
|
|
"--agent-env",
|
|
action="append",
|
|
default=[name for name in os.getenv("RELEASE_AGENT_ENV", "").split(",") if name],
|
|
help="parent environment variable to pass to the coding agent; repeat as needed",
|
|
)
|
|
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,
|
|
agent_env_vars=tuple(args.agent_env),
|
|
)
|
|
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())
|