feat: prove repository rename continuity

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

api/routers/repository_renames.py

@@ -1,12 +1,15 @@
 from __future__ import annotations
 
+import asyncio
 import uuid
 from typing import Any
 
 from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi.encoders import jsonable_encoder
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from api.database import get_session
+from api.events import EventEnvelope, publish_event
 from api.models.repository_rename import RepositoryRenameOperation
 from api.schemas.repository_rename import (
     ForgeIdentityVerifyRequest,
@@ -44,10 +47,77 @@ operation_router = APIRouter(
 )
 
 
+def repository_rename_events(
+    operation: RepositoryRenameOperation,
+    *,
+    outcome: str,
+    no_op: bool = False,
+    verification: dict[str, Any] | None = None,
+) -> list[tuple[str, EventEnvelope]]:
+    """Build credential-free lifecycle events from the durable journal."""
+    telemetry = (operation.evidence or {}).get("telemetry") or {}
+    phase_durations = telemetry.get("phase_durations_ms") or {}
+    attributes = {
+        "operation_id": str(operation.id),
+        "repo_id": str(operation.repo_id),
+        "forge_repository_id": operation.expected_forge_repository_id,
+        "old_slug": operation.old_slug,
+        "new_slug": operation.new_slug,
+        "phase": operation.phase,
+        "actor": operation.actor,
+        "expected_source_commit": operation.expected_source_commit,
+        "outcome": outcome,
+        "no_op": no_op,
+        "phase_duration_ms": int(phase_durations.get(operation.phase) or 0),
+        "retries": int(telemetry.get("retries") or 0),
+        "failures": int(telemetry.get("failures") or 0),
+        "rollback_attempts": int(telemetry.get("rollback_attempts") or 0),
+        "rollback_outcome": telemetry.get("rollback_outcome"),
+        "verification_outcome": (
+            "passed" if verification and verification.get("ok") else
+            "failed" if verification else
+            telemetry.get("verification_outcome", "pending")
+        ),
+        "error_code": operation.error_code if outcome == "failed" else None,
+        "evidence_ref": f"repository-renames/operations/{operation.id}",
+    }
+    subject = (
+        "org.statehub.repo.rename.failed"
+        if outcome == "failed"
+        else "org.statehub.repo.rename.verified"
+        if verification is not None
+        else "org.statehub.repo.rename.rolled_back"
+        if operation.phase == "rolled-back"
+        else "org.statehub.repo.rename.phase"
+    )
+    events = [(subject, EventEnvelope.new(subject, attributes=attributes))]
+    if operation.phase == "completed" and outcome == "succeeded":
+        renamed_subject = "org.statehub.repo.renamed"
+        events.append(
+            (renamed_subject, EventEnvelope.new(renamed_subject, attributes=attributes))
+        )
+    return events
+
+
+def _publish_rename_events(
+    operation: RepositoryRenameOperation,
+    *,
+    outcome: str,
+    no_op: bool = False,
+    verification: dict[str, Any] | None = None,
+) -> None:
+    for subject, envelope in repository_rename_events(
+        operation, outcome=outcome, no_op=no_op, verification=verification
+    ):
+        asyncio.create_task(publish_event(subject, envelope))
+
+
 def _raise(exc: RenameLifecycleError) -> None:
     raise HTTPException(
         status_code=exc.status_code,
-        detail={"code": exc.code, "message": str(exc), "details": exc.details},
+        detail=jsonable_encoder(
+            {"code": exc.code, "message": str(exc), "details": exc.details}
+        ),
     ) from exc
 
 
@@ -129,6 +199,7 @@ async def create_repository_rename_operation(
         operation, no_op = await create_operation(session, gateway, repo_id, body)
     except RenameLifecycleError as exc:
         _raise(exc)
+    _publish_rename_events(operation, outcome="succeeded", no_op=no_op)
     return _operation_read(operation, no_op=no_op)
 
 
@@ -194,7 +265,11 @@ async def apply_repository_rename_phase(
             session, gateway, repo_id, operation_id, phase, body
         )
     except RenameLifecycleError as exc:
+        failed = await session.get(RepositoryRenameOperation, operation_id)
+        if failed is not None:
+            _publish_rename_events(failed, outcome="failed")
         _raise(exc)
+    _publish_rename_events(operation, outcome="succeeded", no_op=no_op)
     return _operation_read(operation, no_op=no_op)
 
 
@@ -210,9 +285,15 @@ async def verify_repository_rename_operation(
 ) -> dict[str, Any]:
     try:
         operation = await load_operation(session, repo_id, operation_id)
-        return await verify_operation(session, gateway, operation)
+        verification = await verify_operation(session, gateway, operation)
     except RenameLifecycleError as exc:
         _raise(exc)
+    _publish_rename_events(
+        operation,
+        outcome="succeeded" if verification["ok"] else "failed",
+        verification=verification,
+    )
+    return verification
 
 
 @router.post(
@@ -236,7 +317,11 @@ async def preflight_repository_rename_rollback(
             confirmation=body.confirmation,
         )
     except RenameLifecycleError as exc:
+        failed = await session.get(RepositoryRenameOperation, operation_id)
+        if failed is not None:
+            _publish_rename_events(failed, outcome="failed")
         _raise(exc)
+    _publish_rename_events(operation, outcome="succeeded")
     return {
         "operation_id": operation.id,
         "repo_id": operation.repo_id,
@@ -266,5 +351,9 @@ async def rollback_repository_rename_operation(
             confirmation=body.confirmation,
         )
     except RenameLifecycleError as exc:
+        failed = await session.get(RepositoryRenameOperation, operation_id)
+        if failed is not None:
+            _publish_rename_events(failed, outcome="failed")
         _raise(exc)
+    _publish_rename_events(operation, outcome="succeeded", no_op=no_op)
     return _operation_read(operation, no_op=no_op)

api/schemas/repository_rename.py

@@ -101,6 +101,7 @@ class RepositoryRenameVerificationRead(BaseModel):
     checks: list[dict[str, Any]]
     baseline_checksum: str
     current_checksum: str
+    relationship_checksums: dict[str, dict[str, str]]
 
 
 class RepositoryRenameRollbackPreflightRead(BaseModel):

api/services/repository_rename.py

@@ -176,23 +176,78 @@ def _record_summary(ids: list[str]) -> dict[str, Any]:
     return {"count": len(ids), "ids": ids, "checksum": checksum(ids)}
 
 
+async def _relationship_rows(
+    session: AsyncSession, query, fields: tuple[str, ...]
+) -> list[dict[str, Any]]:
+    def normalize(value: Any) -> Any:
+        if value is None or isinstance(value, (str, int, float, bool)):
+            return value
+        if isinstance(value, datetime):
+            return value.astimezone(timezone.utc).isoformat()
+        if isinstance(value, uuid.UUID):
+            return str(value)
+        if hasattr(value, "value"):
+            return value.value
+        return str(value)
+
+    result = await session.execute(query)
+    rows = [
+        {field: normalize(getattr(row, field)) for field in fields}
+        for row in result.all()
+    ]
+    return sorted(rows, key=lambda row: canonical_json(row))
+
+
+def _relationship_summary(
+    rows: list[dict[str, Any]], *, totals: dict[str, int] | None = None
+) -> dict[str, Any]:
+    ids = sorted(row["id"] for row in rows)
+    result = {
+        "count": len(rows),
+        "ids": ids,
+        "checksum": checksum(ids),
+        "relationships": rows,
+        "relationship_checksum": checksum(rows),
+    }
+    if totals is not None:
+        result["totals"] = totals
+        result["totals_checksum"] = checksum(totals)
+    return result
+
+
 async def collect_continuity_baseline(
     session: AsyncSession, repo_id: uuid.UUID
 ) -> dict[str, Any]:
-    workplans = await _id_rows(
-        session, select(Workplan.id).where(Workplan.repo_id == repo_id)
-    )
-    workplan_uuids = [uuid.UUID(value) for value in workplans]
-    tasks = await _id_rows(
+    workplans = await _relationship_rows(
         session,
-        select(Task.id).where(
+        select(
+            Workplan.id.label("id"),
+            Workplan.repo_id.label("repo_id"),
+            Workplan.slug.label("slug"),
+        ).where(Workplan.repo_id == repo_id),
+        ("id", "repo_id", "slug"),
+    )
+    workplan_uuids = [uuid.UUID(row["id"]) for row in workplans]
+    tasks = await _relationship_rows(
+        session,
+        select(
+            Task.id.label("id"),
+            Task.workplan_id.label("workplan_id"),
+            Task.record_id.label("record_id"),
+        ).where(
             Task.workplan_id.in_(workplan_uuids) if workplan_uuids else False
         ),
+        ("id", "workplan_id", "record_id"),
     )
-    task_uuids = [uuid.UUID(value) for value in tasks]
-    progress = await _id_rows(
+    task_uuids = [uuid.UUID(row["id"]) for row in tasks]
+    progress = await _relationship_rows(
         session,
-        select(ProgressEvent.id).where(
+        select(
+            ProgressEvent.id.label("id"),
+            ProgressEvent.workplan_id.label("workplan_id"),
+            ProgressEvent.task_id.label("task_id"),
+            ProgressEvent.decision_id.label("decision_id"),
+        ).where(
             or_(
                 ProgressEvent.workplan_id.in_(workplan_uuids)
                 if workplan_uuids
@@ -200,16 +255,31 @@ async def collect_continuity_baseline(
                 ProgressEvent.task_id.in_(task_uuids) if task_uuids else False,
             )
         ),
+        ("id", "workplan_id", "task_id", "decision_id"),
     )
-    decisions = await _id_rows(
+    decisions = await _relationship_rows(
         session,
-        select(Decision.id).where(
+        select(
+            Decision.id.label("id"),
+            Decision.workplan_id.label("workplan_id"),
+        ).where(
             Decision.workplan_id.in_(workplan_uuids) if workplan_uuids else False
         ),
+        ("id", "workplan_id"),
     )
-    token_events = await _id_rows(
+    token_events = await _relationship_rows(
         session,
-        select(TokenEvent.id).where(
+        select(
+            TokenEvent.id.label("id"),
+            TokenEvent.repo_id.label("repo_id"),
+            TokenEvent.workplan_id.label("workplan_id"),
+            TokenEvent.task_id.label("task_id"),
+            TokenEvent.tokens_in.label("tokens_in"),
+            TokenEvent.tokens_out.label("tokens_out"),
+            TokenEvent.cached_input_tokens.label("cached_input_tokens"),
+            TokenEvent.reasoning_output_tokens.label("reasoning_output_tokens"),
+            TokenEvent.raw_total_tokens.label("raw_total_tokens"),
+        ).where(
             or_(
                 TokenEvent.repo_id == repo_id,
                 TokenEvent.workplan_id.in_(workplan_uuids)
@@ -218,44 +288,141 @@ async def collect_continuity_baseline(
                 TokenEvent.task_id.in_(task_uuids) if task_uuids else False,
             )
         ),
+        (
+            "id",
+            "repo_id",
+            "workplan_id",
+            "task_id",
+            "tokens_in",
+            "tokens_out",
+            "cached_input_tokens",
+            "reasoning_output_tokens",
+            "raw_total_tokens",
+        ),
     )
-    sbom_snapshots = await _id_rows(
-        session, select(SBOMSnapshot.id).where(SBOMSnapshot.repo_id == repo_id)
-    )
-    sbom_entries = await _id_rows(
-        session, select(SBOMEntry.id).where(SBOMEntry.repo_id == repo_id)
-    )
-    services = await _id_rows(
+    token_totals = {
+        field: sum(int(row[field]) for row in token_events if row[field] is not None)
+        for field in (
+            "tokens_in",
+            "tokens_out",
+            "cached_input_tokens",
+            "reasoning_output_tokens",
+            "raw_total_tokens",
+        )
+    }
+    sbom_snapshots = await _relationship_rows(
         session,
-        select(ServiceFirstParty.service_id).where(ServiceFirstParty.repo_id == repo_id),
+        select(
+            SBOMSnapshot.id.label("id"),
+            SBOMSnapshot.repo_id.label("repo_id"),
+        ).where(SBOMSnapshot.repo_id == repo_id),
+        ("id", "repo_id"),
     )
-    capabilities = await _id_rows(
+    sbom_entries = await _relationship_rows(
         session,
-        select(CapabilityCatalog.id).where(CapabilityCatalog.repo_id == repo_id),
+        select(
+            SBOMEntry.id.label("id"),
+            SBOMEntry.repo_id.label("repo_id"),
+            SBOMEntry.snapshot_id.label("snapshot_id"),
+        ).where(SBOMEntry.repo_id == repo_id),
+        ("id", "repo_id", "snapshot_id"),
     )
-    interface_changes = await _id_rows(
+    services = await _relationship_rows(
         session,
-        select(InterfaceChange.id).where(InterfaceChange.repo_id == repo_id),
+        select(
+            ServiceFirstParty.service_id.label("id"),
+            ServiceFirstParty.repo_id.label("repo_id"),
+        ).where(ServiceFirstParty.repo_id == repo_id),
+        ("id", "repo_id"),
     )
-    bindings = await _id_rows(
+    capabilities = await _relationship_rows(
         session,
-        select(Workplan.id).where(
+        select(
+            CapabilityCatalog.id.label("id"),
+            CapabilityCatalog.repo_id.label("repo_id"),
+        ).where(CapabilityCatalog.repo_id == repo_id),
+        ("id", "repo_id"),
+    )
+    interface_changes = await _relationship_rows(
+        session,
+        select(
+            InterfaceChange.id.label("id"),
+            InterfaceChange.repo_id.label("repo_id"),
+        ).where(InterfaceChange.repo_id == repo_id),
+        ("id", "repo_id"),
+    )
+    bindings = await _relationship_rows(
+        session,
+        select(
+            Workplan.id.label("id"),
+            Workplan.repo_id.label("repo_id"),
+            Workplan.backing_relative_path.label("backing_relative_path"),
+            Workplan.backing_filename.label("backing_filename"),
+        ).where(
             Workplan.repo_id == repo_id,
             Workplan.backing_relative_path.is_not(None),
         ),
+        ("id", "repo_id", "backing_relative_path", "backing_filename"),
     )
+    slug_rows = await _relationship_rows(
+        session,
+        select(
+            RepositorySlug.id.label("id"),
+            RepositorySlug.repo_id.label("repo_id"),
+            RepositorySlug.slug.label("slug"),
+            RepositorySlug.kind.label("kind"),
+        ).where(RepositorySlug.repo_id == repo_id),
+        ("id", "repo_id", "slug", "kind"),
+    )
+    known_slugs = [row["slug"] for row in slug_rows]
+    messages = await _relationship_rows(
+        session,
+        select(
+            AgentMessage.id.label("id"),
+            AgentMessage.from_agent.label("from_agent"),
+            AgentMessage.to_agent.label("to_agent"),
+            AgentMessage.thread_id.label("thread_id"),
+        ).where(
+            or_(
+                AgentMessage.from_agent.in_(known_slugs),
+                AgentMessage.to_agent.in_(known_slugs),
+            )
+            if known_slugs
+            else False
+        ),
+        ("id", "from_agent", "to_agent", "thread_id"),
+    )
+    active_work = await _active_work(session, repo_id)
+    active_dispatch = [
+        {"id": row["id"], "repo_id": str(repo_id), "kind": "workplan"}
+        for row in active_work["workplans"]
+    ] + [
+        {
+            "id": row["id"],
+            "workplan_id": row["workplan_id"],
+            "kind": "task",
+        }
+        for row in active_work["tasks"]
+    ]
+    active_dispatch = sorted(active_dispatch, key=lambda row: canonical_json(row))
     records = {
-        "workplans": _record_summary(workplans),
-        "tasks": _record_summary(tasks),
-        "progress_events": _record_summary(progress),
-        "decisions": _record_summary(decisions),
-        "token_events": _record_summary(token_events),
-        "sbom_snapshots": _record_summary(sbom_snapshots),
-        "sbom_entries": _record_summary(sbom_entries),
-        "services": _record_summary(services),
-        "capabilities": _record_summary(capabilities),
-        "interface_changes": _record_summary(interface_changes),
-        "workplan_bindings": _record_summary(bindings),
+        "repository": _relationship_summary(
+            [{"id": str(repo_id), "repo_id": str(repo_id)}]
+        ),
+        "workplans": _relationship_summary(workplans),
+        "tasks": _relationship_summary(tasks),
+        "progress_events": _relationship_summary(progress),
+        "decisions": _relationship_summary(decisions),
+        "token_events": _relationship_summary(token_events, totals=token_totals),
+        "sbom_snapshots": _relationship_summary(sbom_snapshots),
+        "sbom_entries": _relationship_summary(sbom_entries),
+        "services": _relationship_summary(services),
+        "capabilities": _relationship_summary(capabilities),
+        "interface_changes": _relationship_summary(interface_changes),
+        "workplan_bindings": _relationship_summary(bindings),
+        "active_dispatch": _relationship_summary(active_dispatch),
+        "aliases": _relationship_summary(slug_rows),
+        "messages": _relationship_summary(messages),
     }
     records["continuity_checksum"] = checksum(records)
     return records
@@ -373,6 +540,11 @@ async def verify_forge_identity(
         raise RenamePreconditionFailed(str(exc)) from exc
     if snapshot is None:
         raise RenamePreconditionFailed("Forge repository is absent or unreadable")
+    if snapshot.name != repo.slug:
+        raise RenamePreconditionFailed(
+            "Forge identity verification followed a repository redirect",
+            details={"expected_name": repo.slug, "actual_name": snapshot.name},
+        )
     if snapshot.repository_id != body.forge_repository_id:
         raise RenamePreconditionFailed(
             "Forge repository ID does not match the asserted immutable ID",
@@ -536,6 +708,15 @@ async def build_preflight(
                 {"code": "forge_source_absent", "message": "Canonical Forge repository is absent or unreadable"}
             )
         else:
+            if old_snapshot.name != repo.slug:
+                blockers.append(
+                    {
+                        "code": "forge_redirected",
+                        "message": "Forge source lookup resolved to another coordinate",
+                        "expected": repo.slug,
+                        "actual": old_snapshot.name,
+                    }
+                )
             if old_snapshot.repository_id != identity.forge_repository_id:
                 blockers.append(
                     {
@@ -786,6 +967,15 @@ async def create_operation(
             "phases": {
                 "preflighted": {"at": now.isoformat(), "resumed": False}
             },
+            "telemetry": {
+                "started_at": now.isoformat(),
+                "phase_attempts": {"preflighted": 1},
+                "phase_durations_ms": {"preflighted": 0},
+                "retries": 0,
+                "failures": 0,
+                "rollback_attempts": 0,
+                "verification_outcome": "pending",
+            },
         },
     )
     session.add(operation)
@@ -860,7 +1050,12 @@ async def list_operations(
     return list(result.scalars().all())
 
 
-def _assert_snapshot(operation: RepositoryRenameOperation, snapshot: ForgeRepositorySnapshot) -> None:
+def _assert_snapshot(
+    operation: RepositoryRenameOperation,
+    snapshot: ForgeRepositorySnapshot,
+    *,
+    expected_name: str | None = None,
+) -> None:
     if snapshot.repository_id != operation.expected_forge_repository_id:
         raise RenamePreconditionFailed(
             "Forge repository ID changed",
@@ -881,6 +1076,11 @@ def _assert_snapshot(operation: RepositoryRenameOperation, snapshot: ForgeReposi
         )
     if not snapshot.projection_readable or not snapshot.projection_source_present:
         raise RenamePreconditionFailed("Forge work-record projection is unreadable")
+    if expected_name is not None and snapshot.name != expected_name:
+        raise RenamePreconditionFailed(
+            "Forge response resolved to a different repository coordinate",
+            details={"expected_name": expected_name, "actual_name": snapshot.name},
+        )
 
 
 async def _forge_at(
@@ -904,13 +1104,13 @@ async def _apply_forge_rename(
     old = await _forge_at(gateway, operation, operation.old_slug)
     new = await _forge_at(gateway, operation, operation.new_slug)
     if new is not None:
-        _assert_snapshot(operation, new)
+        _assert_snapshot(operation, new, expected_name=operation.new_slug)
         if old is not None:
             raise RenamePreconditionFailed("Both old and new Forge names are claimed")
         return {"resumed": True, "forge": new.as_dict()}
     if old is None:
         raise RenamePreconditionFailed("Forge repository is absent at both expected names")
-    _assert_snapshot(operation, old)
+    _assert_snapshot(operation, old, expected_name=operation.old_slug)
     try:
         renamed = await gateway.rename(
             instance=operation.expected_forge_instance,
@@ -920,7 +1120,7 @@ async def _apply_forge_rename(
         )
     except (ForgeRepositoryUnreadable, ForgeRepositoryConflict) as exc:
         raise RenamePreconditionFailed(str(exc)) from exc
-    _assert_snapshot(operation, renamed)
+    _assert_snapshot(operation, renamed, expected_name=operation.new_slug)
     return {"resumed": False, "forge": renamed.as_dict()}
 
 
@@ -1005,6 +1205,7 @@ async def verify_operation(
     else:
         add("forge_readable", forge is not None, True, forge is not None)
     if forge is not None:
+        add("forge_coordinate", forge.name == expected_name, expected_name, forge.name)
         add("forge_repository_id", forge.repository_id == operation.expected_forge_repository_id, operation.expected_forge_repository_id, forge.repository_id)
         add("source_commit", forge.head_commit == operation.expected_source_commit, operation.expected_source_commit, forge.head_commit)
         add("default_branch", forge.default_branch == operation.expected_default_branch, operation.expected_default_branch, forge.default_branch)
@@ -1022,6 +1223,8 @@ async def verify_operation(
     baseline = (operation.evidence.get("preflight") or {}).get("baselines") or {}
     expected_checksum = baseline.get("continuity_checksum", "")
     missing: dict[str, list[str]] = {}
+    changed_relationships: dict[str, list[str]] = {}
+    preserved_relationship_checksums: dict[str, str] = {}
     counts: dict[str, dict[str, int]] = {}
     for record_type, baseline_record in baseline.items():
         if record_type == "continuity_checksum" or not isinstance(baseline_record, dict):
@@ -1030,6 +1233,30 @@ async def verify_operation(
         absent = sorted(set(baseline_record.get("ids") or []) - set(current_record.get("ids") or []))
         if absent:
             missing[record_type] = absent
+        if record_type != "aliases":
+            baseline_rows = {
+                row["id"]: row
+                for row in baseline_record.get("relationships") or []
+            }
+            current_rows = {
+                row["id"]: row
+                for row in current_record.get("relationships") or []
+            }
+            changed = sorted(
+                record_id
+                for record_id, baseline_row in baseline_rows.items()
+                if current_rows.get(record_id) != baseline_row
+            )
+            if changed:
+                changed_relationships[record_type] = changed
+            preserved_rows = [
+                current_rows[record_id]
+                for record_id in baseline_rows
+                if record_id in current_rows
+            ]
+            preserved_relationship_checksums[record_type] = checksum(
+                sorted(preserved_rows, key=lambda row: canonical_json(row))
+            )
         counts[record_type] = {
             "baseline": int(baseline_record.get("count") or 0),
             "current": int(current_record.get("count") or 0),
@@ -1037,13 +1264,54 @@ async def verify_operation(
     # New telemetry may legitimately arrive while a phased operation is being
     # executed.  Continuity means every baseline identity still exists; it does
     # not freeze the repository's append-only history at the preflight count.
-    add("relationship_continuity", not missing, {"missing": {}}, {"missing": missing})
+    add(
+        "relationship_continuity",
+        not missing and not changed_relationships,
+        {"missing": {}, "changed": {}},
+        {"missing": missing, "changed": changed_relationships},
+    )
     add(
         "record_counts_non_decreasing",
         all(item["current"] >= item["baseline"] for item in counts.values()),
         {name: item["baseline"] for name, item in counts.items()},
         {name: item["current"] for name, item in counts.items()},
     )
+    baseline_token_totals = (baseline.get("token_events") or {}).get("totals") or {}
+    current_token_totals = (current.get("token_events") or {}).get("totals") or {}
+    add(
+        "token_totals_non_decreasing",
+        all(
+            int(current_token_totals.get(name) or 0) >= int(value or 0)
+            for name, value in baseline_token_totals.items()
+        ),
+        baseline_token_totals,
+        current_token_totals,
+    )
+    slug_result = await session.execute(
+        select(RepositorySlug).where(
+            RepositorySlug.repo_id == operation.repo_id,
+            RepositorySlug.slug.in_((operation.old_slug, operation.new_slug)),
+        )
+    )
+    slug_routes = {row.slug: row.kind for row in slug_result.scalars().all()}
+    if operation.phase in {"preflighted", "forge-renamed"}:
+        expected_routes = {operation.old_slug: "canonical"}
+    elif operation.phase == "rolled-back":
+        expected_routes = {
+            operation.old_slug: "canonical",
+            operation.new_slug: "alias",
+        }
+    else:
+        expected_routes = {
+            operation.old_slug: "alias",
+            operation.new_slug: "canonical",
+        }
+    add(
+        "slug_routes",
+        all(slug_routes.get(slug) == kind for slug, kind in expected_routes.items()),
+        expected_routes,
+        slug_routes,
+    )
     return {
         "operation_id": operation.id,
         "repo_id": operation.repo_id,
@@ -1052,12 +1320,63 @@ async def verify_operation(
         "checks": checks,
         "baseline_checksum": expected_checksum,
         "current_checksum": current["continuity_checksum"],
+        "relationship_checksums": {
+            "baseline": {
+                name: value.get("relationship_checksum")
+                for name, value in baseline.items()
+                if isinstance(value, dict) and name != "aliases"
+            },
+            "preserved": preserved_relationship_checksums,
+        },
     }
 
 
+def _telemetry(operation: RepositoryRenameOperation) -> tuple[dict[str, Any], dict[str, Any]]:
+    journal = deepcopy(operation.evidence or {})
+    telemetry = dict(journal.get("telemetry") or {})
+    telemetry.setdefault("started_at", operation.preflighted_at.isoformat())
+    telemetry.setdefault("phase_attempts", {})
+    telemetry.setdefault("phase_durations_ms", {})
+    telemetry.setdefault("retries", 0)
+    telemetry.setdefault("failures", 0)
+    telemetry.setdefault("rollback_attempts", 0)
+    telemetry.setdefault("verification_outcome", "pending")
+    return journal, telemetry
+
+
+def _record_retry(operation: RepositoryRenameOperation, phase: str) -> None:
+    journal, telemetry = _telemetry(operation)
+    attempts = dict(telemetry["phase_attempts"])
+    attempts[phase] = int(attempts.get(phase) or 0) + 1
+    telemetry["phase_attempts"] = attempts
+    telemetry["retries"] = int(telemetry["retries"]) + 1
+    telemetry["last_retry"] = {"phase": phase, "at": utcnow().isoformat()}
+    journal["telemetry"] = telemetry
+    operation.evidence = journal
+
+
+def _record_failure(
+    operation: RepositoryRenameOperation, phase: str, exc: RenameLifecycleError
+) -> None:
+    journal, telemetry = _telemetry(operation)
+    failures_by_phase = dict(telemetry.get("failures_by_phase") or {})
+    failures_by_phase[phase] = int(failures_by_phase.get(phase) or 0) + 1
+    telemetry["failures_by_phase"] = failures_by_phase
+    telemetry["failures"] = int(telemetry["failures"]) + 1
+    telemetry["last_failure"] = {
+        "phase": phase,
+        "code": exc.code,
+        "at": utcnow().isoformat(),
+    }
+    if phase.startswith("rollback"):
+        telemetry["rollback_outcome"] = "failed"
+    journal["telemetry"] = telemetry
+    operation.evidence = journal
+
+
 def _record_phase(operation: RepositoryRenameOperation, phase: str, evidence: dict[str, Any]) -> None:
     now = utcnow()
-    journal = deepcopy(operation.evidence or {})
+    journal, telemetry = _telemetry(operation)
     phases = dict(journal.get("phases") or {})
     # Verification payloads use native UUID/datetime values for API response
     # typing.  The operation journal is JSONB, so normalize at the boundary
@@ -1065,6 +1384,26 @@ def _record_phase(operation: RepositoryRenameOperation, phase: str, evidence: di
     safe_evidence = json.loads(canonical_json(evidence))
     phases[phase] = {"at": now.isoformat(), **safe_evidence}
     journal["phases"] = phases
+    attempts = dict(telemetry["phase_attempts"])
+    attempts[phase] = int(attempts.get(phase) or 0) + 1
+    telemetry["phase_attempts"] = attempts
+    durations = dict(telemetry["phase_durations_ms"])
+    durations[phase] = max(
+        0, int((now - operation.phase_changed_at).total_seconds() * 1000)
+    )
+    telemetry["phase_durations_ms"] = durations
+    telemetry["last_phase"] = phase
+    if phase == "rollback-preflight":
+        telemetry["rollback_attempts"] = int(telemetry["rollback_attempts"]) + 1
+        telemetry["rollback_outcome"] = "approved"
+    elif phase == "rolled-back":
+        telemetry["rollback_outcome"] = "completed"
+    verification = evidence.get("verification")
+    if isinstance(verification, dict):
+        telemetry["verification_outcome"] = (
+            "passed" if verification.get("ok") else "failed"
+        )
+    journal["telemetry"] = telemetry
     operation.evidence = journal
     operation.phase = phase
     operation.phase_changed_at = now
@@ -1101,6 +1440,9 @@ async def apply_phase(
     current_index = _phase_index(operation.phase)
     requested_index = _phase_index(requested_phase)
     if current_index >= requested_index >= 0:
+        _record_retry(operation, requested_phase)
+        await session.commit()
+        await session.refresh(operation)
         return operation, True
     if operation.phase != body.expected_phase:
         raise RenamePreconditionFailed(
@@ -1119,17 +1461,35 @@ async def apply_phase(
             forge = await _forge_at(gateway, operation, operation.new_slug)
             if forge is None:
                 raise RenamePreconditionFailed("Forge rename is not observable")
-            _assert_snapshot(operation, forge)
+            _assert_snapshot(operation, forge, expected_name=operation.new_slug)
             evidence = await _apply_statehub_rebind(session, operation)
         elif requested_phase == "source-synced":
-            if not body.evidence:
+            if not body.evidence or body.evidence.get("fresh_clone") is not True:
                 raise RenamePreconditionFailed(
-                    "Source synchronization requires operator evidence"
+                    "Source synchronization requires fresh-clone operator evidence"
+                )
+            clone_repository_id = body.evidence.get("forge_repository_id")
+            if clone_repository_id != operation.expected_forge_repository_id:
+                raise RenamePreconditionFailed(
+                    "Fresh clone points at the wrong Forge repository ID",
+                    details={
+                        "expected": operation.expected_forge_repository_id,
+                        "actual": clone_repository_id,
+                    },
+                )
+            clone_head = body.evidence.get("head_commit")
+            if clone_head != operation.expected_source_commit:
+                raise RenamePreconditionFailed(
+                    "Fresh clone head does not match the rename baseline",
+                    details={
+                        "expected": operation.expected_source_commit,
+                        "actual": clone_head,
+                    },
                 )
             forge = await _forge_at(gateway, operation, operation.new_slug)
             if forge is None:
                 raise RenamePreconditionFailed("Renamed Forge repository is absent")
-            _assert_snapshot(operation, forge)
+            _assert_snapshot(operation, forge, expected_name=operation.new_slug)
             evidence = {"forge": forge.as_dict(), "operator_evidence": body.evidence}
         elif requested_phase == "consumers-verified":
             if not body.checks or not all(body.checks.values()):
@@ -1161,9 +1521,10 @@ async def apply_phase(
         # phase is precisely what lets the next request discover and resume it.
         await session.rollback()
         failed = await load_operation(session, repo_id, operation_id, for_update=True)
+        _record_failure(failed, requested_phase, exc)
         failed.error_code = exc.code
         failed.error_message = str(exc)
-        failed.error_details = exc.details
+        failed.error_details = json.loads(canonical_json(exc.details))
         failed.error_at = utcnow()
         await session.commit()
         raise
@@ -1173,9 +1534,10 @@ async def apply_phase(
             "Repository rename phase lost a database compare-and-set race"
         )
         failed = await load_operation(session, repo_id, operation_id, for_update=True)
+        _record_failure(failed, requested_phase, failure)
         failed.error_code = failure.code
         failed.error_message = str(failure)
-        failed.error_details = failure.details
+        failed.error_details = json.loads(canonical_json(failure.details))
         failed.error_at = utcnow()
         await session.commit()
         raise failure from exc
@@ -1225,7 +1587,7 @@ async def rollback_preflight(
     if operation.phase == "preflighted":
         if old is not None and new is None:
             try:
-                _assert_snapshot(operation, old)
+                _assert_snapshot(operation, old, expected_name=operation.old_slug)
             except RenameLifecycleError as exc:
                 blockers.append({"code": exc.code, "message": str(exc), **exc.details})
         elif old is None and new is not None:
@@ -1233,7 +1595,7 @@ async def rollback_preflight(
             # principal interruption case the operation ID must recover from.
             rollback_from_phase = "forge-renamed-unrecorded"
             try:
-                _assert_snapshot(operation, new)
+                _assert_snapshot(operation, new, expected_name=operation.new_slug)
             except RenameLifecycleError as exc:
                 blockers.append({"code": exc.code, "message": str(exc), **exc.details})
         elif old is None:
@@ -1247,7 +1609,7 @@ async def rollback_preflight(
             blockers.append({"code": "renamed_forge_repository_absent"})
         else:
             try:
-                _assert_snapshot(operation, new)
+                _assert_snapshot(operation, new, expected_name=operation.new_slug)
             except RenameLifecycleError as exc:
                 blockers.append({"code": exc.code, "message": str(exc), **exc.details})
     irreversible = [
@@ -1262,7 +1624,16 @@ async def rollback_preflight(
         "irreversible": irreversible,
     }
     if blockers:
-        raise RenamePreconditionFailed("Repository rename rollback is unsafe", details=data)
+        failure = RenamePreconditionFailed(
+            "Repository rename rollback is unsafe", details=data
+        )
+        _record_failure(operation, "rollback-preflight", failure)
+        operation.error_code = failure.code
+        operation.error_message = str(failure)
+        operation.error_details = json.loads(canonical_json(failure.details))
+        operation.error_at = utcnow()
+        await session.commit()
+        raise failure
     journal = deepcopy(operation.evidence or {})
     journal["rollback_preflight"] = data
     operation.evidence = journal
@@ -1319,6 +1690,9 @@ async def apply_rollback(
 ) -> tuple[RepositoryRenameOperation, bool]:
     operation = await load_operation(session, repo_id, operation_id, for_update=True)
     if operation.phase == "rolled-back":
+        _record_retry(operation, "rolled-back")
+        await session.commit()
+        await session.refresh(operation)
         return operation, True
     if operation.phase != expected_phase or operation.phase != "rollback-preflight":
         raise RenamePreconditionFailed(
@@ -1330,35 +1704,56 @@ async def apply_rollback(
     rollback_from = ((operation.evidence or {}).get("rollback_preflight") or {}).get("rollback_from_phase")
     if not rollback_from:
         raise RenamePreconditionFailed("Rollback preflight evidence is missing")
-    old = await _forge_at(gateway, operation, operation.old_slug)
-    new = await _forge_at(gateway, operation, operation.new_slug)
-    forge_resumed = False
-    if rollback_from != "preflighted":
-        if old is not None:
-            _assert_snapshot(operation, old)
-            if new is not None:
-                raise RenamePreconditionFailed("Both Forge coordinates are claimed during rollback")
-            forge_resumed = True
-        else:
-            if new is None:
-                raise RenamePreconditionFailed("Forge repository is absent during rollback")
-            _assert_snapshot(operation, new)
-            try:
-                restored = await gateway.rename(
-                    instance=operation.expected_forge_instance,
-                    owner=operation.expected_forge_owner,
-                    old_name=operation.new_slug,
-                    new_name=operation.old_slug,
+    try:
+        old = await _forge_at(gateway, operation, operation.old_slug)
+        new = await _forge_at(gateway, operation, operation.new_slug)
+        forge_resumed = False
+        if rollback_from != "preflighted":
+            if old is not None:
+                _assert_snapshot(operation, old, expected_name=operation.old_slug)
+                if new is not None:
+                    raise RenamePreconditionFailed(
+                        "Both Forge coordinates are claimed during rollback"
+                    )
+                forge_resumed = True
+            else:
+                if new is None:
+                    raise RenamePreconditionFailed(
+                        "Forge repository is absent during rollback"
+                    )
+                _assert_snapshot(operation, new, expected_name=operation.new_slug)
+                try:
+                    restored = await gateway.rename(
+                        instance=operation.expected_forge_instance,
+                        owner=operation.expected_forge_owner,
+                        old_name=operation.new_slug,
+                        new_name=operation.old_slug,
+                    )
+                except (ForgeRepositoryUnreadable, ForgeRepositoryConflict) as exc:
+                    raise RenamePreconditionFailed(str(exc)) from exc
+                _assert_snapshot(
+                    operation, restored, expected_name=operation.old_slug
                 )
-            except (ForgeRepositoryUnreadable, ForgeRepositoryConflict) as exc:
-                raise RenamePreconditionFailed(str(exc)) from exc
-            _assert_snapshot(operation, restored)
-    statehub = await _rollback_statehub(session, operation)
-    _record_phase(
-        operation,
-        "rolled-back",
-        {"rollback_from_phase": rollback_from, "forge_resumed": forge_resumed, "statehub": statehub},
-    )
-    await session.commit()
-    await session.refresh(operation)
-    return operation, False
+        statehub = await _rollback_statehub(session, operation)
+        _record_phase(
+            operation,
+            "rolled-back",
+            {
+                "rollback_from_phase": rollback_from,
+                "forge_resumed": forge_resumed,
+                "statehub": statehub,
+            },
+        )
+        await session.commit()
+        await session.refresh(operation)
+        return operation, False
+    except RenameLifecycleError as exc:
+        await session.rollback()
+        failed = await load_operation(session, repo_id, operation_id, for_update=True)
+        _record_failure(failed, "rolled-back", exc)
+        failed.error_code = exc.code
+        failed.error_message = str(exc)
+        failed.error_details = json.loads(canonical_json(exc.details))
+        failed.error_at = utcnow()
+        await session.commit()
+        raise