G11 / I-4: continuous off-site mirror, and a namespace bug fixed

I-4 said 'never a one-way door' and had no implementation. Now it does.

  - ensure the GitHub counterpart exists (idempotent), then ask Gitea to keep
    it in step with sync_on_commit=True. An hourly timer means an hour of work
    can be the thing you lose, and that window is invisible until it costs you.
  - mirror status reports what is TRUE including 'we do not know'. An
    unconfigured mirror reports unconfigured, NEVER healthy — same posture as
    me-fleet.ts refusing to say 'online' when it only knows 'registered'.
  - lag past the threshold is a P2, not a shrug. A mirror nobody checks is a
    belief, not a backup, and this ecosystem already lost 37 days to a canary
    everyone assumed was fine.

Gitea owns the replication rather than a hand-rolled loop, because a background
job that fails silently is exactly how the registry's integrity refresh spent
its entire life calling a 404 and incrementing a counter instead of raising.

Also fixes a real bug I had written myself: list_versions derived the Gitea
namespace from the CALLER, which is correct only while the caller is the owner
and addresses the wrong namespace the moment a collaborator asks — surfacing as
'not found', which is the hardest kind of bug to see. Now derived from the repo,
with a test that keeps it that way.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

api/app/routes/repos.py

@@ -31,6 +31,8 @@ from api.app.errors import RepairPointer
 from api.app.models.core import (
     CreatedVia,
     GrantRole,
+    Mirror,
+    MirrorState,
     Repo,
     RepoGrant,
     RepoState,
@@ -39,6 +41,7 @@ from api.app.models.core import (
     Visibility,
 )
 from api.app.services.gitea_client import GiteaClient
+from api.app.services.mirror import MirrorService
 
 router = APIRouter(prefix="/api/v1/repos", tags=["repos"])
 
@@ -111,6 +114,18 @@ def _sessionmaker(request: Request) -> async_sessionmaker[AsyncSession]:
     return maker
 
 
+def _repo_owner_login(repo: Repo) -> str:
+    """The owning login for a repo as STORED, not as inferred from the caller.
+
+    Deriving this from the caller works only while the caller is the owner, and
+    silently addresses the wrong namespace the moment a collaborator calls. That
+    class of bug reads as "not found" and is very hard to see.
+    """
+    if repo.passport:
+        return f"agent-{repo.passport.lower().replace('-', '')}"
+    return f"u-{repo.identity_id[:24]}"
+
+
 def _owner_login(caller: Caller) -> str:
     """One namespace rule for humans and agents alike (I-6)."""
     if caller.actor_type == ActorType.agent and caller.passport:
@@ -272,11 +287,14 @@ async def list_versions(
     """
     async with _sessionmaker(request)() as session:
         repo = await _load_repo(session, repo_id, caller)
-        owner = _owner_login(caller) if repo.passport == caller.passport else None
+        # Derive the namespace from the REPO, never from the caller: the
+        # caller-derived form is right only while the caller is the owner, and
+        # addresses the wrong namespace the moment a collaborator asks. It then
+        # surfaces as "not found", which is about the hardest bug to see.
+        owner, slug, display = _repo_owner_login(repo), repo.slug, repo.display_name
 
     gitea = GiteaClient(request.app.state.settings)
-    owner = owner or f"u-{repo.identity_id[:24]}"
-    commits = await gitea.list_commits(owner, repo.slug)
+    commits = await gitea.list_commits(owner, slug)
 
     versions = [
         {
@@ -294,12 +312,12 @@ async def list_versions(
         # "There are 1 saved versions" is the kind of sloppiness the vocabulary
         # law exists to catch. Copy is design material, not decoration (I-9).
         "speak": (
-            f"'{repo.display_name}' has 1 saved version. You can go back to it."
+            f"'{display}' has 1 saved version. You can go back to it."
             if len(versions) == 1
-            else f"'{repo.display_name}' has {len(versions)} saved versions. "
+            else f"'{display}' has {len(versions)} saved versions. "
             "You can go back to any of them."
             if versions
-            else f"'{repo.display_name}' is empty so far."
+            else f"'{display}' is empty so far."
         ),
     }
 
@@ -438,3 +456,65 @@ async def get_repo(
         "recorded_versions": len(versions),
         "state": repo.state.value,
     }
+
+
+# --------------------------------------------------------------------------
+# I-4 / G11 — the off-site copy
+# --------------------------------------------------------------------------
+@router.post("/{repo_id}/mirror", status_code=201)
+async def enable_mirror(
+    repo_id: uuid.UUID,
+    request: Request,
+    caller: Annotated[Caller, Depends(get_caller)],
+) -> dict:
+    """Turn on the continuous off-site copy.
+
+    Deliberately idempotent and deliberately loud on failure: a mirror that
+    quietly stopped working is worse than no mirror, because it is a backup you
+    believe in.
+    """
+    settings = request.app.state.settings
+    async with _sessionmaker(request)() as session:
+        repo = await _load_repo(session, repo_id, caller)
+        owner = _repo_owner_login(repo)
+        display, slug = repo.display_name, repo.slug
+        private = repo.visibility != Visibility.public
+
+    mirror = MirrorService(settings)
+    remote = await mirror.ensure_github_repo(slug, display, private)
+    await mirror.attach_push_mirror(owner, slug, remote)
+
+    async with _sessionmaker(request)() as session:
+        session.add(
+            Mirror(repo_id=repo_id, remote_url=remote, direction="push", state=MirrorState.healthy)
+        )
+        await session.commit()
+
+    return {
+        "remote": remote,
+        "sync_on_commit": True,
+        "speak": "A second copy of this project is now kept somewhere else, automatically.",
+        "state_proof": {"remote": remote},
+        "next_actions": ["windy_git.mirror_status"],
+    }
+
+
+@router.get("/{repo_id}/mirror")
+async def mirror_status(
+    repo_id: uuid.UUID,
+    request: Request,
+    caller: Annotated[Caller, Depends(get_caller)],
+) -> dict:
+    async with _sessionmaker(request)() as session:
+        repo = await _load_repo(session, repo_id, caller)
+        owner, slug = _repo_owner_login(repo), repo.slug
+
+    status = await MirrorService(request.app.state.settings).status(owner, slug)
+    speak = {
+        "healthy": "A second copy of this project is up to date.",
+        "degraded": "The second copy is behind. Your work here is safe.",
+        "absent": "There is no second copy of this project yet.",
+        "unconfigured": "Off-site copies aren't switched on yet.",
+        "unknown": "We can't tell how the second copy is doing right now.",
+    }[status["state"]]
+    return {**status, "speak": speak}