feat(13-08): implement create-folder and rename collision retry, stale guard, and reconciliation

- create_cloud_folder: bounded collision retry (up to 5 attempts) with keep_both_name counter suffix (D-05, D-06)
- create_cloud_folder: stale precondition marks parent folder non-fresh and returns typed stale result (D-07)
- rename_cloud_item: stale precondition marks parent folder non-fresh via folder state update (D-07)
- create_cloud_folder: reconcile-before-return upserts new folder in cloud_items and invalidates parent (T-13-26)
- rename_cloud_item: reconcile-before-return upserts renamed item and invalidates parent folder (T-13-26)
- Import upsert_cloud_item, update_folder_state, keep_both_name, CloudResource from service modules

backend/api/cloud/operations.py

@@ -47,9 +47,15 @@ from deps.auth import get_regular_user
 from deps.db import get_db
 from deps.utils import get_client_ip
 from services.audit import write_audit_log
-from services.cloud_items import ConnectionNotFound
+from services.cloud_items import (
+    ConnectionNotFound,
+    update_folder_state,
+    upsert_cloud_item,
+)
+from services.cloud_operations import keep_both_name
 from services.rate_limiting import account_limiter
 from storage.cloud_base import (
+    CloudResource,
     MUT_KIND_CONFLICT,
     MUT_KIND_DELETED,
     MUT_KIND_FOLDER,
@@ -70,6 +76,9 @@ from storage.cloud_base import (
     PreviewSupport,
 )
 
+# Maximum collision retry attempts for create-folder (D-06)
+_CREATE_FOLDER_MAX_RETRIES = 5
+
 router = APIRouter()
 
 
@@ -375,10 +384,14 @@ async def create_cloud_folder(
     """Create a folder in connected cloud storage.
 
     D-05: Name collision auto-suffix: 'Projects (1)', 'Projects (2)', etc.
-    D-06: Concurrent collision race is handled by the adapter with bounded retry.
+    D-06: Bounded retry up to _CREATE_FOLDER_MAX_RETRIES when a concurrent client
+          takes the candidate name between check and mutation.
+    D-07: Stale precondition returned by the provider stops the mutation, marks the
+          parent folder as non-fresh, and returns typed {kind: 'stale'} for retry.
 
     Returns {kind: 'folder', provider_item_id, name, parent_ref} on success.
-    Returns typed error body for unsupported providers, auth failures, etc.
+    After success, the new folder is upserted into cloud_items and the parent folder
+    state is invalidated before the response is returned (reconcile-before-return).
 
     T-13-01: Owner-scoped. T-13-14: No credentials in response.
     """
@@ -386,31 +399,90 @@ async def create_cloud_folder(
 
     conn, adapter = await _resolve_and_get_adapter(session, connection_id, current_user.id)
 
-    result = await adapter.create_folder(
+    candidate_name = body.name
-        parent_ref=body.parent_ref,
+    last_result: dict = {}
-        name=body.name,
-        connection_id=connection_id,
-        user_id=current_user.id,
-    )
 
-    kind = result.get("kind")
+    for attempt in range(_CREATE_FOLDER_MAX_RETRIES + 1):
+        result = await adapter.create_folder(
+            parent_ref=body.parent_ref,
+            name=candidate_name,
+            connection_id=connection_id,
+            user_id=current_user.id,
+        )
+        last_result = result
+        kind = result.get("kind")
 
-    if kind == MUT_KIND_FOLDER:
+        if kind == MUT_KIND_FOLDER:
-        return {
+            # Success: reconcile before returning (T-13-24, T-13-26)
-            "kind": "folder",
+            provider_item_id = result.get("provider_item_id")
-            "provider_item_id": result.get("provider_item_id"),
+            resolved_name = result.get("name", candidate_name)
-            "name": result.get("name", body.name),
+            resolved_parent_ref = result.get("parent_ref", body.parent_ref)
-            "parent_ref": result.get("parent_ref", body.parent_ref),
+
-        }
+            if provider_item_id:
+                resource = CloudResource(
+                    id=uuid.uuid4(),
+                    provider_item_id=provider_item_id,
+                    connection_id=connection_id,
+                    user_id=current_user.id,
+                    name=resolved_name,
+                    kind="folder",
+                    parent_ref=resolved_parent_ref,
+                )
+                # Upsert gives the new folder stable row identity in cloud_items
+                await upsert_cloud_item(session, user_id=str(current_user.id), resource=resource)
+
+                # Invalidate parent folder so next browse triggers a provider re-list
+                invalidate_parent = resolved_parent_ref if resolved_parent_ref is not None else ""
+                await update_folder_state(
+                    session,
+                    user_id=str(current_user.id),
+                    connection_id=str(connection_id),
+                    parent_ref=invalidate_parent,
+                    refresh_state="warning",
+                    error_code="create_folder_mutated",
+                    error_message="Folder contents changed by create-folder — re-listing required.",
+                )
+
+                await session.commit()
+
+            return {
+                "kind": "folder",
+                "provider_item_id": provider_item_id,
+                "name": resolved_name,
+                "parent_ref": resolved_parent_ref,
+            }
+
+        if kind == MUT_KIND_CONFLICT and attempt < _CREATE_FOLDER_MAX_RETRIES:
+            # D-05/D-06: auto-suffix with counter and retry
+            candidate_name = keep_both_name(body.name, attempt + 1)
+            continue
+
+        if kind == MUT_KIND_STALE:
+            # D-07: Stop mutation, mark parent folder as non-fresh, return typed stale result
+            stale_parent = body.parent_ref if body.parent_ref is not None else ""
+            await update_folder_state(
+                session,
+                user_id=str(current_user.id),
+                connection_id=str(connection_id),
+                parent_ref=stale_parent,
+                refresh_state="warning",
+                error_code="stale_listing",
+                error_message="Parent folder changed externally — re-listing required before retry.",
+            )
+            await session.commit()
+            return _mutation_error_response(result)
+
+        # All other non-success kinds (offline, reauth, unsupported, conflict exhausted)
+        break
 
     # Unsupported operation — return typed body instead of 500
-    if kind == MUT_KIND_UNSUPPORTED:
+    if last_result.get("kind") == MUT_KIND_UNSUPPORTED:
         return {
             "kind": "unsupported_operation",
             "reason": "provider_unsupported",
         }
 
-    return _mutation_error_response(result)
+    return _mutation_error_response(last_result)
 
 
 # ── Rename ────────────────────────────────────────────────────────────────────
@@ -428,9 +500,14 @@ async def rename_cloud_item(
 ) -> dict:
     """Rename a cloud file or folder.
 
-    D-05: Counter-suffix collision policy — auto-suffixes on name collision.
+    D-05: Rename collision is surfaced to the user (they chose the name explicitly).
-    D-07: etag guards against operating on stale metadata; returns
+    D-07: etag guards against operating on stale metadata. On mismatch:
-          {kind: 'stale', reason: 'item_changed'} on mismatch (HTTP 409).
+          stops mutation, marks parent folder as non-fresh, returns
+          {kind: 'stale', reason: 'item_changed'} (HTTP 409) so the frontend
+          re-lists before retrying.
+
+    On success: upserts the updated item name in cloud_items and invalidates the
+    parent folder state before returning (reconcile-before-return, T-13-26).
 
     Returns {kind: 'renamed', provider_item_id, name} on success.
 
@@ -449,12 +526,92 @@ async def rename_cloud_item(
     kind = result.get("kind")
 
     if kind == MUT_KIND_UPDATED:
+        # Success: reconcile the updated name in cloud_items before returning (T-13-26)
+        resolved_provider_item_id = result.get("provider_item_id", item_id)
+        resolved_name = result.get("name", body.new_name)
+
+        # Resolve the parent_ref from the existing CloudItem so we can invalidate it
+        from sqlalchemy import select
+        from db.models import CloudItem
+
+        existing_result = await session.execute(
+            select(CloudItem).where(
+                CloudItem.connection_id == connection_id,
+                CloudItem.provider_item_id == item_id,
+                CloudItem.user_id == current_user.id,
+                CloudItem.deleted_at.is_(None),
+            )
+        )
+        existing_item = existing_result.scalar_one_or_none()
+        item_parent_ref = existing_item.parent_ref if existing_item else None
+        item_kind = existing_item.kind if existing_item else "file"
+        item_content_type = existing_item.content_type if existing_item else None
+        item_size = existing_item.provider_size if existing_item else None
+
+        # Upsert the item with the new name — uses provider_item_id as stable identity key
+        resource = CloudResource(
+            id=uuid.uuid4(),
+            provider_item_id=resolved_provider_item_id,
+            connection_id=connection_id,
+            user_id=current_user.id,
+            name=resolved_name,
+            kind=item_kind,
+            parent_ref=item_parent_ref,
+            content_type=item_content_type,
+            size=item_size,
+        )
+        await upsert_cloud_item(session, user_id=str(current_user.id), resource=resource)
+
+        # Invalidate the parent folder so the next browse reflects the renamed item
+        invalidate_parent = item_parent_ref if item_parent_ref is not None else ""
+        await update_folder_state(
+            session,
+            user_id=str(current_user.id),
+            connection_id=str(connection_id),
+            parent_ref=invalidate_parent,
+            refresh_state="warning",
+            error_code="rename_mutated",
+            error_message="Folder contents changed by rename — re-listing required.",
+        )
+
+        await session.commit()
+
         return {
             "kind": "renamed",
-            "provider_item_id": result.get("provider_item_id", item_id),
+            "provider_item_id": resolved_provider_item_id,
-            "name": result.get("name", body.new_name),
+            "name": resolved_name,
         }
 
+    if kind == MUT_KIND_STALE:
+        # D-07: Stop mutation, mark parent folder as non-fresh, return typed stale result.
+        # Resolve parent_ref from existing CloudItem metadata if available.
+        from sqlalchemy import select
+        from db.models import CloudItem
+
+        stale_result = await session.execute(
+            select(CloudItem).where(
+                CloudItem.connection_id == connection_id,
+                CloudItem.provider_item_id == item_id,
+                CloudItem.user_id == current_user.id,
+                CloudItem.deleted_at.is_(None),
+            )
+        )
+        stale_item = stale_result.scalar_one_or_none()
+        stale_parent = stale_item.parent_ref if stale_item else None
+        stale_parent_ref = stale_parent if stale_parent is not None else ""
+
+        await update_folder_state(
+            session,
+            user_id=str(current_user.id),
+            connection_id=str(connection_id),
+            parent_ref=stale_parent_ref,
+            refresh_state="warning",
+            error_code="stale_listing",
+            error_message="Item changed externally — re-listing required before retry.",
+        )
+        await session.commit()
+        return _mutation_error_response(result)
+
     return _mutation_error_response(result)