From 9dabc57d9af5a3c311a0ad68d59635925e019707 Mon Sep 17 00:00:00 2001 From: opencode-agent Date: Thu, 6 Aug 2026 15:25:42 +0000 Subject: [PATCH] feat(oracles): add Forgejo backend to status scripts --- .opencode/scripts/pipeline-status.py | 133 ++++++++++++++++++++++++++- .opencode/scripts/project-status.py | 47 +++++++++- .opencode/scripts/spec-status.py | 44 ++++++++- 3 files changed, 219 insertions(+), 5 deletions(-) diff --git a/.opencode/scripts/pipeline-status.py b/.opencode/scripts/pipeline-status.py index 43aff15..27fe2de 100644 --- a/.opencode/scripts/pipeline-status.py +++ b/.opencode/scripts/pipeline-status.py @@ -28,6 +28,8 @@ import re import subprocess import sys import time +import urllib.error +import urllib.request from dataclasses import dataclass from enum import StrEnum from pathlib import Path @@ -85,10 +87,135 @@ class PhaseResult: detail: str +def _forgejo_request(method: str, path: str, body: dict | None = None) -> tuple[int, str, str]: + """Forgejo REST API call. Returns (status_code, body_text, error). + + On non-2xx returns (status_code, body_text, ""). ``urlopen`` raises + HTTPError for non-2xx which carries the body; we surface it so callers + can branch on the HTTP status. Network/parse errors return (0, "", err). + """ + base = os.environ.get("FORGEJO_URL") + token = os.environ.get("FORGEJO_TOKEN") + headers = {"Authorization": f"token {token}", "Accept": "application/json"} + data = None + if body is not None: + headers["Content-Type"] = "application/json" + data = json.dumps(body).encode() + req = urllib.request.Request( # noqa: S310 - base URL is operator-configured + f"{base}/api/v1{path}", method=method, headers=headers, data=data + ) + try: + with urllib.request.urlopen(req) as r: # noqa: S310 - base URL is operator-configured + return r.status, r.read().decode("utf-8", "replace"), "" + except urllib.error.HTTPError as e: + return e.code, e.read().decode("utf-8", "replace"), "" + except OSError as e: + return 0, "", str(e) + + +def _forgejo_ci_rollup(repo: str, sha: str) -> tuple[int, str, str]: + """Build a GitHub-style statusCheckRollup from the Forgejo commit status.""" + sc, text, err = _forgejo_request("GET", f"/repos/{repo}/commits/{sha}/status") + if sc != 200: + return 1, "", err or f"commit status HTTP {sc}" + combined = json.loads(text) + rollup: list[dict] = [] + for s in combined.get("statuses", []): + st = s.get("status", "").lower() + if st == "success": + rollup.append({"status": "COMPLETED", "conclusion": "SUCCESS"}) + elif st == "pending": + rollup.append({"status": "IN_PROGRESS", "conclusion": ""}) + elif st in ("failure", "error"): + rollup.append({"status": "COMPLETED", "conclusion": "FAILURE"}) + else: + rollup.append({"status": "QUEUED", "conclusion": ""}) + return 0, json.dumps({"statusCheckRollup": rollup}), "" + + +def _forgejo_pr_view(repo: str, n: str, fields: str) -> tuple[int, str, str]: # noqa: PLR0911 + """Translate ``gh pr view N --json `` to Forgejo API calls.""" + rc, pr_text, err = _forgejo_request("GET", f"/repos/{repo}/pulls/{n}") + if rc != 200: + return 1, "", err or f"PR API HTTP {rc}" + pr = json.loads(pr_text) + if fields == "statusCheckRollup": + return _forgejo_ci_rollup(repo, pr["head"]["sha"]) + if fields == "comments": + sc, comments_text, ce = _forgejo_request("GET", f"/repos/{repo}/issues/{n}/comments") + if sc != 200: + return 1, "", ce or f"comments HTTP {sc}" + comments = json.loads(comments_text) + return 0, json.dumps({"comments": [{"body": c.get("body", "")} for c in comments]}), "" + if fields == "state": + state = "MERGED" if pr.get("merged") else pr.get("state", "").upper() + return 0, json.dumps({"state": state}), "" + if fields: + return 0, json.dumps({fields: pr.get(fields, "")}), "" + return 0, pr_text, "" + + +def _forgejo_pr_dispatch(repo: str, args: list[str]) -> tuple[int, str, str]: + """Translate ``gh pr ...`` argv to Forgejo API calls.""" + action = args[1] if len(args) > 1 else "" + num = args[2] if len(args) > 2 and args[2].lstrip("-").isdigit() else None + if action == "view" and num is not None: + fields_idx = args.index("--json") if "--json" in args else -1 + fields = args[fields_idx + 1] if fields_idx >= 0 else "" + return _forgejo_pr_view(repo, num, fields) + if action == "list": + sc, out_text, e = _forgejo_request("GET", f"/repos/{repo}/pulls?state=open") + if sc != 200: + return 1, "", e or f"pr list HTTP {sc}" + pulls = json.loads(out_text) + return 0, json.dumps([{"number": p["number"]} for p in pulls]), "" + return 1, "", f"gh pr argv {args!r} not supported in Forgejo mode" + + +def _forgejo_gh_dispatch(args: list[str]) -> tuple[int, str, str] | None: # noqa: PLR0911 + """Translate a ``gh`` argv to a Forgejo API call. Returns None to defer. + + Returns ``(rc, stdout, stderr)`` shaped like ``run_cmd`` so callers stay + unchanged. Returns ``None`` if ``FORGEJO_URL`` is unset (defer to gh) or + the argv is not a supported gh subcommand. + """ + if not os.environ.get("FORGEJO_URL"): + return None + if not os.environ.get("FORGEJO_TOKEN"): + return 1, "", "Forgejo mode requires FORGEJO_TOKEN" + if not args or args[0] != "gh": + return None + repo_idx = args.index("--repo") if "--repo" in args else -1 + repo = args[repo_idx + 1] if repo_idx >= 0 else None + if repo is None: + return 1, "", "Forgejo mode requires --repo owner/name" + sub = args[1] if len(args) > 1 else "" + if sub == "auth" and "status" in args: + return 0, "", "" + if sub == "pr": + return _forgejo_pr_dispatch(repo, args) + if sub == "issue" and len(args) > 2 and args[2].lstrip("-").isdigit(): + sc, out_text, e = _forgejo_request("GET", f"/repos/{repo}/issues/{args[2]}") + if sc == 200: + return 0, out_text, "" + return 1, "", e or f"issue HTTP {sc}" + return 1, "", f"gh argv {args!r} not supported in Forgejo mode" + + def run_cmd(args: list[str]) -> tuple[int, str, str]: - """Run a command, return (returncode, stdout, stderr).""" - result = subprocess.run(args, capture_output=True, text=True, check=False) - return result.returncode, result.stdout, result.stderr + """Run a command, return (returncode, stdout, stderr). + + Forgejo dispatch (ADR-forgejo): when ``FORGEJO_URL`` is set, ``gh`` argv + is translated to a Forgejo REST API call instead of spawning ``gh``. + GitHub users (no ``FORGEJO_URL``) see byte-identical behaviour — the + ``gh`` / ``git`` subprocess path is untouched. + """ + if args and args[0] == "gh": + result = _forgejo_gh_dispatch(args) + if result is not None: + return result + proc = subprocess.run(args, capture_output=True, text=True, check=False) + return proc.returncode, proc.stdout, proc.stderr @dataclass(frozen=True) diff --git a/.opencode/scripts/project-status.py b/.opencode/scripts/project-status.py index c631419..429aabb 100644 --- a/.opencode/scripts/project-status.py +++ b/.opencode/scripts/project-status.py @@ -37,10 +37,13 @@ import argparse import ast import importlib.util import json +import os import re import subprocess import sys import tomllib +import urllib.error +import urllib.request from dataclasses import dataclass, field from enum import StrEnum from pathlib import Path @@ -187,8 +190,50 @@ class RepoCtx: # ── helpers ────────────────────────────────────────────────────────────────── +def _forgejo_get(path: str) -> tuple[int, str, str]: + """Forgejo REST GET. Returns (status_code, body_text, error).""" + base = os.environ.get("FORGEJO_URL") + token = os.environ.get("FORGEJO_TOKEN") + req = urllib.request.Request( # noqa: S310 - operator-configured base URL + f"{base}/api/v1{path}", + headers={"Authorization": f"token {token}", "Accept": "application/json"}, + ) + try: + with urllib.request.urlopen(req) as r: # noqa: S310 - operator-configured base URL + return r.status, r.read().decode("utf-8", "replace"), "" + except urllib.error.HTTPError as e: + return e.code, e.read().decode("utf-8", "replace"), "" + except OSError as e: + return 0, "", str(e) + + def run_cmd(args: list[str]) -> tuple[int, str, str]: - """Run a command, return (returncode, stdout, stderr). Read-only intent.""" + """Run a command, return (returncode, stdout, stderr). Read-only intent. + + Forgejo dispatch: when ``FORGEJO_URL`` is set, ``gh api + repos//rules/branches/main`` is routed to the Forgejo + ``branch_protections/main`` endpoint (different schema, but the + downstream string-matching checks ``"pull_request"`` / + ``"required_status_checks"`` which Forgejo's BranchProtection fields + contain). GitHub users (no ``FORGEJO_URL``) see byte-identical behaviour. + """ + if ( + args + and args[0] == "gh" + and os.environ.get("FORGEJO_URL") + and len(args) >= 2 + and args[1] == "api" + ): + if not os.environ.get("FORGEJO_TOKEN"): + return 1, "", "Forgejo mode requires FORGEJO_TOKEN" + m = re.match(r"repos/([^/]+/[^/]+)/rules/branches/(\S+)", " ".join(args[2:])) + if m: + repo, branch = m.group(1), m.group(2) + sc, out_text, err = _forgejo_get(f"/repos/{repo}/branch_protections/{branch}") + if sc == 200: + return 0, out_text, "" + return 1, "", err or f"branch protection HTTP {sc}" + return 1, "", f"gh api argv {args!r} not supported in Forgejo mode" result = subprocess.run(args, capture_output=True, text=True, check=False) return result.returncode, result.stdout, result.stderr diff --git a/.opencode/scripts/spec-status.py b/.opencode/scripts/spec-status.py index 70c7e94..8be100e 100644 --- a/.opencode/scripts/spec-status.py +++ b/.opencode/scripts/spec-status.py @@ -38,9 +38,12 @@ from __future__ import annotations import functools import importlib.util +import os import re import subprocess import sys +import urllib.error +import urllib.request from dataclasses import dataclass from enum import StrEnum from pathlib import Path @@ -111,8 +114,47 @@ class PhaseResult: detail: str +def _forgejo_get(path: str) -> tuple[int, str, str]: + """Forgejo REST GET. Returns (status_code, body_text, error).""" + base = os.environ.get("FORGEJO_URL") + token = os.environ.get("FORGEJO_TOKEN") + req = urllib.request.Request( # noqa: S310 - operator-configured base URL + f"{base}/api/v1{path}", + headers={"Authorization": f"token {token}", "Accept": "application/json"}, + ) + try: + with urllib.request.urlopen(req) as r: # noqa: S310 - operator-configured base URL + return r.status, r.read().decode("utf-8", "replace"), "" + except urllib.error.HTTPError as e: + return e.code, e.read().decode("utf-8", "replace"), "" + except OSError as e: + return 0, "", str(e) + + def run_cmd(args: list[str]) -> tuple[int, str, str]: - """Run a command, return (returncode, stdout, stderr).""" + """Run a command, return (returncode, stdout, stderr). + + Forgejo dispatch: when ``FORGEJO_URL`` is set, ``gh issue view`` is routed + to the Forgejo REST API instead of spawning ``gh``. GitHub users (no + ``FORGEJO_URL``) see byte-identical behaviour — the gh/git subprocess path + is untouched. + """ + if args and args[0] == "gh" and os.environ.get("FORGEJO_URL"): + if not os.environ.get("FORGEJO_TOKEN"): + return 1, "", "Forgejo mode requires FORGEJO_TOKEN" + repo_idx = args.index("--repo") if "--repo" in args else -1 + repo = args[repo_idx + 1] if repo_idx >= 0 else None + if ( + args[1:3] == ["issue", "view"] + and repo + and len(args) > 2 + and args[2].lstrip("-").isdigit() + ): + sc, out_text, err = _forgejo_get(f"/repos/{repo}/issues/{args[2]}") + if sc == 200: + return 0, out_text, "" + return 1, "", err or f"issue HTTP {sc}" + return 1, "", f"gh argv {args!r} not supported in Forgejo mode" result = subprocess.run(args, capture_output=True, text=True, check=False) return result.returncode, result.stdout, result.stderr