Recover external repository renames through verified Forgejo redirects

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a070b5-4994-7271-bd8b-7c3dbcedec4b

api/services/repository_rename.py

@@ -11,6 +11,7 @@ from copy import deepcopy
 from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
 from typing import Any
+from urllib.parse import quote
 
 from sqlalchemy import or_, select
 from sqlalchemy.exc import IntegrityError
@@ -44,6 +45,7 @@ from api.schemas.repository_rename import (
 from api.services.forge_repository import (
     ForgeRepositoryConflict,
     ForgeRepositoryGateway,
+    ForgeRepositoryRedirected,
     ForgeRepositorySnapshot,
     ForgeRepositoryUnreadable,
 )
@@ -1101,13 +1103,36 @@ async def _forge_at(
 async def _apply_forge_rename(
     gateway: ForgeRepositoryGateway, operation: RepositoryRenameOperation
 ) -> dict[str, Any]:
-    old = await _forge_at(gateway, operation, operation.old_slug)
     new = await _forge_at(gateway, operation, operation.new_slug)
     if new is not None:
+        # A redirect is compatibility evidence only. Establish immutable identity
+        # at the requested new coordinate before accepting the old-name alias.
         _assert_snapshot(operation, new, expected_name=operation.new_slug)
+        try:
+            old = await gateway.inspect(
+                instance=operation.expected_forge_instance,
+                owner=operation.expected_forge_owner,
+                name=operation.old_slug,
+            )
+        except ForgeRepositoryRedirected as exc:
+            canonical_api = (
+                operation.expected_forge_instance.rstrip("/")
+                + "/api/v1/repos/"
+                + quote(operation.expected_forge_owner, safe="")
+                + "/"
+                + quote(operation.new_slug, safe="")
+            )
+            if exc.target_url != canonical_api:
+                raise RenamePreconditionFailed(
+                    "Old Forge name redirects outside the verified new coordinate"
+                ) from exc
+            old = None
+        except ForgeRepositoryUnreadable as exc:
+            raise RenamePreconditionFailed(str(exc)) from exc
         if old is not None:
             raise RenamePreconditionFailed("Both old and new Forge names are claimed")
         return {"resumed": True, "forge": new.as_dict()}
+    old = await _forge_at(gateway, operation, operation.old_slug)
     if old is None:
         raise RenamePreconditionFailed("Forge repository is absent at both expected names")
     _assert_snapshot(operation, old, expected_name=operation.old_slug)