feat: add repository rename lifecycle API

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a049a4-ee9f-78e1-9d66-2cb0f9bea3e3

api/services/forge_repository.py

@@ -0,0 +1,203 @@
+"""Narrow Forgejo repository boundary used by repository rename operations.
+
+The lifecycle service depends on this protocol instead of calling Forgejo
+directly.  Tests replace it with an in-memory gateway, which makes interruption
+and resume tests deterministic and guarantees that API tests cannot rename a
+live repository.
+"""
+
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Protocol
+from urllib.parse import quote
+
+import httpx
+
+from api.config import settings
+from api.services.forge_credential import forge_read_token
+
+
+WRITE_TOKEN_ENV = "FORGE_RENAME_TOKEN"
+WRITE_TOKEN_FILE_ENV = "FORGE_RENAME_TOKEN_FILE"
+
+
+class ForgeRepositoryError(RuntimeError):
+    pass
+
+
+class ForgeRepositoryUnreadable(ForgeRepositoryError):
+    pass
+
+
+class ForgeRepositoryConflict(ForgeRepositoryError):
+    pass
+
+
+@dataclass(frozen=True)
+class ForgeRepositorySnapshot:
+    repository_id: int
+    owner: str
+    name: str
+    full_name: str
+    default_branch: str
+    head_commit: str
+    clone_url: str | None
+    html_url: str | None
+    projection_readable: bool
+    projection_source_present: bool
+    projection_entry_count: int | None
+
+    def as_dict(self) -> dict:
+        return {
+            "repository_id": self.repository_id,
+            "owner": self.owner,
+            "name": self.name,
+            "full_name": self.full_name,
+            "default_branch": self.default_branch,
+            "head_commit": self.head_commit,
+            "clone_url": self.clone_url,
+            "html_url": self.html_url,
+            "projection_readable": self.projection_readable,
+            "projection_source_present": self.projection_source_present,
+            "projection_entry_count": self.projection_entry_count,
+        }
+
+
+class ForgeRepositoryGateway(Protocol):
+    async def inspect(
+        self, *, instance: str, owner: str, name: str
+    ) -> ForgeRepositorySnapshot | None: ...
+
+    async def rename(
+        self, *, instance: str, owner: str, old_name: str, new_name: str
+    ) -> ForgeRepositorySnapshot: ...
+
+
+def _write_token() -> str | None:
+    path = os.environ.get(WRITE_TOKEN_FILE_ENV)
+    if path:
+        try:
+            return Path(path).read_text(encoding="utf-8").strip() or None
+        except OSError:
+            return None
+    return (os.environ.get(WRITE_TOKEN_ENV) or "").strip() or None
+
+
+class ForgejoRepositoryGateway:
+    """Forgejo v1 API adapter with separate read and rename credentials."""
+
+    @staticmethod
+    def _headers(token: str | None) -> dict[str, str]:
+        return {"Authorization": f"token {token}"} if token else {}
+
+    @staticmethod
+    def _api(instance: str, owner: str, name: str) -> str:
+        base = instance.rstrip("/")
+        return f"{base}/api/v1/repos/{quote(owner, safe='')}/{quote(name, safe='')}"
+
+    async def _inspect_with_token(
+        self, *, instance: str, owner: str, name: str, token: str | None
+    ) -> ForgeRepositorySnapshot | None:
+        url = self._api(instance, owner, name)
+        timeout = settings.repository_rename_forge_timeout_seconds
+        try:
+            async with httpx.AsyncClient(timeout=timeout) as client:
+                response = await client.get(url, headers=self._headers(token))
+                if response.status_code == 404:
+                    if token is None:
+                        raise ForgeRepositoryUnreadable(
+                            "Forge returned not-found without an authenticated absence proof"
+                        )
+                    return None
+                response.raise_for_status()
+                repo = response.json()
+                branch_name = str(repo.get("default_branch") or "")
+                if not branch_name:
+                    raise ForgeRepositoryUnreadable("Forge repository has no default branch")
+                branch = await client.get(
+                    f"{url}/branches/{quote(branch_name, safe='')}",
+                    headers=self._headers(token),
+                )
+                branch.raise_for_status()
+                branch_data = branch.json()
+                commit = branch_data.get("commit") or {}
+                head = str(commit.get("id") or commit.get("sha") or "")
+                if not head:
+                    raise ForgeRepositoryUnreadable("Forge default branch has no readable head")
+                projection = await client.get(
+                    f"{url}/contents/workplans",
+                    params={"ref": head},
+                    headers=self._headers(token),
+                )
+                source_present = projection.status_code == 200
+                if projection.status_code not in {200, 404}:
+                    projection.raise_for_status()
+                entries = projection.json() if source_present else None
+        except ForgeRepositoryError:
+            raise
+        except (httpx.HTTPError, ValueError, KeyError) as exc:
+            raise ForgeRepositoryUnreadable(
+                f"Forge repository inspection failed ({type(exc).__name__})"
+            ) from exc
+        return ForgeRepositorySnapshot(
+            repository_id=int(repo["id"]),
+            owner=str((repo.get("owner") or {}).get("login") or owner),
+            name=str(repo["name"]),
+            full_name=str(repo.get("full_name") or f"{owner}/{name}"),
+            default_branch=branch_name,
+            head_commit=head,
+            clone_url=repo.get("clone_url"),
+            html_url=repo.get("html_url"),
+            projection_readable=True,
+            projection_source_present=source_present,
+            projection_entry_count=(len(entries) if isinstance(entries, list) else None),
+        )
+
+    async def inspect(
+        self, *, instance: str, owner: str, name: str
+    ) -> ForgeRepositorySnapshot | None:
+        return await self._inspect_with_token(
+            instance=instance, owner=owner, name=name, token=forge_read_token()
+        )
+
+    async def rename(
+        self, *, instance: str, owner: str, old_name: str, new_name: str
+    ) -> ForgeRepositorySnapshot:
+        token = _write_token()
+        if not token:
+            raise ForgeRepositoryUnreadable("Forge rename credential is unavailable")
+        url = self._api(instance, owner, old_name)
+        try:
+            async with httpx.AsyncClient(
+                timeout=settings.repository_rename_forge_timeout_seconds
+            ) as client:
+                response = await client.patch(
+                    url,
+                    json={"name": new_name},
+                    headers=self._headers(token),
+                )
+                if response.status_code in {409, 422}:
+                    raise ForgeRepositoryConflict("Forge rejected the target repository name")
+                response.raise_for_status()
+        except ForgeRepositoryError:
+            raise
+        except httpx.HTTPError as exc:
+            raise ForgeRepositoryUnreadable(
+                f"Forge repository rename failed ({type(exc).__name__})"
+            ) from exc
+        renamed = await self._inspect_with_token(
+            instance=instance, owner=owner, name=new_name, token=token
+        )
+        if renamed is None:
+            raise ForgeRepositoryUnreadable("Renamed Forge repository is not readable")
+        return renamed
+
+
+_gateway = ForgejoRepositoryGateway()
+
+
+def get_forge_repository_gateway() -> ForgeRepositoryGateway:
+    return _gateway