feat(oracles): add Forgejo backend to status scripts

.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 <fields>`` 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)."""
+    """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
+    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)

.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/<repo>/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
 

.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