test(13-08): add RED tests for create-folder and rename collision retry, stale guard, and reconciliation

- Test 1: collision auto-retry with bounded window (D-05, D-06)
- Test 2: stale guard updates folder state and returns typed stale kind (D-07)
- Test 3: create-folder and rename success upserts CloudItem and invalidates parent folder
- Test 4: failed rename does not mutate CloudItem rows

backend/tests/test_cloud_mutations.py

@@ -1229,3 +1229,575 @@ async def test_upload_failed_does_not_mutate_cloud_items(async_client, db_sessio
         f"Failed upload must not create CloudItem rows (Plan 06 Task 1 behavior 3): "
         f"count changed {before_count} → {after_count}"
     )
+
+
+# ── Phase 13 Plan 08 Task 1 (RED): Collision retry, stale guard, provider normalization ─
+
+
+async def test_create_folder_collision_triggers_bounded_retry_and_succeeds(async_client, db_session):
+    """POST create-folder with collision auto-retries with counter suffix and returns success.
+
+    Plan 08 Task 1 behavior 1 (D-05, D-06): When the adapter returns a collision on
+    the first attempt, the service layer must retry with the next counter-suffixed name
+    (e.g. 'Projects (1)', 'Projects (2)') within a bounded window. The final response
+    must be kind='folder' with the non-colliding name — never a bare 'conflict' error
+    returned to the client when the retry window has not been exhausted.
+
+    FAILS: Current create-folder route returns conflict without retrying.
+    """
+    from storage.cloud_base import (
+        MUT_KIND_CONFLICT, MUT_KIND_FOLDER,
+        MUT_REASON_NAME_COLLISION, MUT_REASON_CREATED,
+    )
+    import uuid as _uuid2
+
+    auth = await _create_user_and_token(db_session)
+    conn = await _create_cloud_connection(db_session, auth["user"].id)
+
+    # First call collides; second call (with counter suffix) succeeds
+    collision_then_success = [
+        {"kind": MUT_KIND_CONFLICT, "reason": MUT_REASON_NAME_COLLISION},
+        {
+            "kind": MUT_KIND_FOLDER,
+            "reason": MUT_REASON_CREATED,
+            "provider_item_id": str(_uuid2.uuid4()),
+            "name": "Projects (1)",
+            "parent_ref": None,
+        },
+    ]
+    mock_adapter = _make_mock_mutable_adapter()
+    mock_adapter.create_folder = AsyncMock(side_effect=collision_then_success)
+
+    payload = {"parent_ref": None, "name": "Projects"}
+
+    with patch("storage.cloud_backend_factory.build_mutable_cloud_adapter", return_value=mock_adapter):
+        resp = await async_client.post(
+            f"/api/cloud/connections/{conn.id}/folders",
+            headers=auth["headers"],
+            json=payload,
+        )
+
+    assert resp.status_code == 201, (
+        f"Expected 201 after bounded retry succeeds, got {resp.status_code}: {resp.text}"
+    )
+    body = resp.json()
+    assert body.get("kind") == "folder", (
+        f"Expected kind='folder' after collision retry, got {body.get('kind')!r}"
+    )
+    assert "Projects (1)" in body.get("name", ""), (
+        f"Expected counter-suffixed name after retry, got {body.get('name')!r}"
+    )
+
+
+async def test_rename_stale_triggers_folder_refresh_and_returns_stale_kind(async_client, db_session):
+    """PATCH rename with stale etag refreshes parent folder state and returns kind='stale'.
+
+    Plan 08 Task 1 behavior 2 (D-07): Stale etag mismatch must not force the mutation.
+    Instead: stop mutation, mark the parent folder state as stale (non-fresh), and
+    return typed {kind: 'stale', reason: 'item_changed'} so the frontend knows to
+    re-list before retrying.
+
+    FAILS: Current rename route returns stale kind but does NOT update folder state.
+    """
+    from storage.cloud_base import MUT_KIND_STALE, MUT_REASON_ITEM_CHANGED
+    from db.models import CloudFolderState
+    from sqlalchemy import select
+
+    auth = await _create_user_and_token(db_session)
+    conn = await _create_cloud_connection(db_session, auth["user"].id)
+
+    # Pre-seed a CloudItem so we have a known parent_ref
+    from db.models import CloudItem
+    test_parent_ref = "stale_test_parent"
+    existing_item = CloudItem(
+        id=_uuid.uuid4(),
+        user_id=auth["user"].id,
+        connection_id=conn.id,
+        provider_item_id="stale_item_ref",
+        name="Old Name.pdf",
+        kind="file",
+        parent_ref=test_parent_ref,
+    )
+    db_session.add(existing_item)
+    await db_session.commit()
+
+    mock_adapter = _make_mock_mutable_adapter(
+        rename_result={"kind": MUT_KIND_STALE, "reason": MUT_REASON_ITEM_CHANGED}
+    )
+    payload = {"new_name": "New Name.pdf", "etag": "stale-etag-v0"}
+
+    with patch("storage.cloud_backend_factory.build_mutable_cloud_adapter", return_value=mock_adapter):
+        resp = await async_client.patch(
+            f"/api/cloud/connections/{conn.id}/items/stale_item_ref/rename",
+            headers=auth["headers"],
+            json=payload,
+        )
+
+    # Must return stale kind at HTTP 409
+    assert resp.status_code == 409, (
+        f"Expected 409 for stale rename, got {resp.status_code}: {resp.text}"
+    )
+    body = resp.json()
+    assert body.get("kind") == "stale", f"Expected kind='stale', got {body.get('kind')!r}"
+    assert body.get("reason") == "item_changed", (
+        f"Expected reason='item_changed', got {body.get('reason')!r}"
+    )
+
+    # Parent folder must be marked as non-fresh after stale detection
+    result = await db_session.execute(
+        select(CloudFolderState).where(
+            CloudFolderState.connection_id == conn.id,
+            CloudFolderState.parent_ref == test_parent_ref,
+        )
+    )
+    fs = result.scalar_one_or_none()
+    assert fs is not None, (
+        "Stale rename must create/update a CloudFolderState row for the parent folder "
+        "(Plan 08 Task 1 behavior 2 — folder refresh required before retry)"
+    )
+    assert fs.refresh_state != "fresh", (
+        f"Stale rename must mark parent folder as non-fresh; got refresh_state={fs.refresh_state!r}"
+    )
+
+
+async def test_create_folder_collision_exhausts_retries_returns_conflict(async_client, db_session):
+    """POST create-folder exhausting all retry attempts returns typed kind='conflict'.
+
+    Plan 08 Task 1 behavior 1 (D-06): Bounded retry must stop after N attempts.
+    When all counter-suffixed names also collide, the response must be a typed
+    {kind: 'conflict', reason: 'name_collision'} — never an unhandled exception or 500.
+
+    FAILS: Current implementation does not retry and always returns conflict on first collision.
+    """
+    from storage.cloud_base import MUT_KIND_CONFLICT, MUT_REASON_NAME_COLLISION
+
+    auth = await _create_user_and_token(db_session)
+    conn = await _create_cloud_connection(db_session, auth["user"].id)
+
+    # All attempts collide — adapter always returns conflict
+    mock_adapter = _make_mock_mutable_adapter(
+        create_folder_result={"kind": MUT_KIND_CONFLICT, "reason": MUT_REASON_NAME_COLLISION}
+    )
+
+    payload = {"parent_ref": None, "name": "AlwaysConflict"}
+
+    with patch("storage.cloud_backend_factory.build_mutable_cloud_adapter", return_value=mock_adapter):
+        resp = await async_client.post(
+            f"/api/cloud/connections/{conn.id}/folders",
+            headers=auth["headers"],
+            json=payload,
+        )
+
+    # After bounded retry exhausted, return conflict — not 500
+    assert resp.status_code != 500, (
+        "Exhausted collision retries must not produce a 500"
+    )
+    body = resp.json()
+    assert body.get("kind") in ("conflict", "folder"), (
+        f"Expected kind 'conflict' or 'folder' after all retries exhausted, got {body.get('kind')!r}"
+    )
+
+
+async def test_rename_collision_returns_conflict_kind(async_client, db_session):
+    """PATCH rename with a colliding name returns typed kind='conflict'.
+
+    Plan 08 Task 1 behavior 1 (D-05): Rename collision must return a typed conflict
+    body. Unlike create-folder (which auto-retries with a counter suffix), rename
+    collision is surfaced to the user — they chose the new name explicitly.
+
+    The route must not hide the collision or silently choose a different name.
+    """
+    from storage.cloud_base import MUT_KIND_CONFLICT, MUT_REASON_NAME_COLLISION
+
+    auth = await _create_user_and_token(db_session)
+    conn = await _create_cloud_connection(db_session, auth["user"].id)
+
+    mock_adapter = _make_mock_mutable_adapter(
+        rename_result={"kind": MUT_KIND_CONFLICT, "reason": MUT_REASON_NAME_COLLISION}
+    )
+    payload = {"new_name": "Existing Report.pdf", "etag": "v1"}
+
+    with patch("storage.cloud_backend_factory.build_mutable_cloud_adapter", return_value=mock_adapter):
+        resp = await async_client.patch(
+            f"/api/cloud/connections/{conn.id}/items/fake_item_id/rename",
+            headers=auth["headers"],
+            json=payload,
+        )
+
+    assert resp.status_code in (200, 409), (
+        f"Rename conflict must produce 200 or 409, got {resp.status_code}: {resp.text}"
+    )
+    body = resp.json()
+    assert body.get("kind") == "conflict", (
+        f"Expected kind='conflict' for rename collision, got {body.get('kind')!r}"
+    )
+    assert body.get("reason") == "name_collision", (
+        f"Expected reason='name_collision', got {body.get('reason')!r}"
+    )
+
+
+async def test_create_folder_stale_returns_stale_kind_and_refreshes_folder(async_client, db_session):
+    """POST create-folder stale precondition returns kind='stale' and marks folder non-fresh.
+
+    Plan 08 Task 1 behavior 2 (D-07): If a provider returns stale on create-folder
+    (precondition check), the route must stop the mutation, mark the parent folder
+    state as non-fresh, and return typed {kind: 'stale'} so the frontend re-lists.
+
+    FAILS: Current create-folder route does not update folder state on stale.
+    """
+    from storage.cloud_base import MUT_KIND_STALE, MUT_REASON_ITEM_CHANGED
+    from db.models import CloudFolderState
+    from sqlalchemy import select
+
+    auth = await _create_user_and_token(db_session)
+    conn = await _create_cloud_connection(db_session, auth["user"].id)
+
+    stale_parent_ref = "stale_folder_parent"
+    mock_adapter = _make_mock_mutable_adapter(
+        create_folder_result={"kind": MUT_KIND_STALE, "reason": MUT_REASON_ITEM_CHANGED}
+    )
+
+    payload = {"parent_ref": stale_parent_ref, "name": "NewFolder"}
+
+    with patch("storage.cloud_backend_factory.build_mutable_cloud_adapter", return_value=mock_adapter):
+        resp = await async_client.post(
+            f"/api/cloud/connections/{conn.id}/folders",
+            headers=auth["headers"],
+            json=payload,
+        )
+
+    assert resp.status_code != 500, "Stale create-folder must not produce a 500"
+    body = resp.json()
+    assert body.get("kind") in ("stale", "conflict", "error"), (
+        f"Expected typed stale/conflict kind, got {body.get('kind')!r}"
+    )
+
+    # Parent folder must be marked as non-fresh after stale detection
+    result = await db_session.execute(
+        select(CloudFolderState).where(
+            CloudFolderState.connection_id == conn.id,
+            CloudFolderState.parent_ref == stale_parent_ref,
+        )
+    )
+    fs = result.scalar_one_or_none()
+    assert fs is not None, (
+        "Stale create-folder must create/update a CloudFolderState row for the parent folder"
+    )
+    assert fs.refresh_state != "fresh", (
+        f"Stale create-folder must mark parent as non-fresh; got {fs.refresh_state!r}"
+    )
+
+
+# ── Phase 13 Plan 08 Task 2 (RED): Reconcile create-folder and rename through stable identity ─
+
+
+async def test_create_folder_success_upserts_cloud_item_before_returning(async_client, db_session):
+    """POST create-folder success upserts the new folder into cloud_items before returning.
+
+    Plan 08 Task 2 behavior 1: Successful create-folder must update navigation metadata
+    through centralized reconciliation (cloud_items.upsert_cloud_item) before the route
+    returns success. The returned 'folder' body's provider_item_id must have a matching
+    CloudItem row by the time the response is received.
+
+    FAILS: Current create-folder route does not call upsert_cloud_item on success.
+    """
+    from storage.cloud_base import MUT_KIND_FOLDER, MUT_REASON_CREATED
+    from db.models import CloudItem
+    from sqlalchemy import select
+    import uuid as _uuid2
+
+    auth = await _create_user_and_token(db_session)
+    conn = await _create_cloud_connection(db_session, auth["user"].id)
+
+    new_folder_id = str(_uuid2.uuid4())
+    mock_adapter = _make_mock_mutable_adapter(
+        create_folder_result={
+            "kind": MUT_KIND_FOLDER,
+            "reason": MUT_REASON_CREATED,
+            "provider_item_id": new_folder_id,
+            "name": "My New Folder",
+            "parent_ref": "root_folder_ref",
+        }
+    )
+
+    payload = {"parent_ref": "root_folder_ref", "name": "My New Folder"}
+
+    with patch("storage.cloud_backend_factory.build_mutable_cloud_adapter", return_value=mock_adapter):
+        resp = await async_client.post(
+            f"/api/cloud/connections/{conn.id}/folders",
+            headers=auth["headers"],
+            json=payload,
+        )
+
+    assert resp.status_code == 201, f"Expected 201 for folder creation, got {resp.status_code}"
+    body = resp.json()
+    assert body.get("kind") == "folder"
+
+    # After successful create-folder, the provider_item_id must have a CloudItem row
+    result = await db_session.execute(
+        select(CloudItem).where(
+            CloudItem.connection_id == conn.id,
+            CloudItem.provider_item_id == new_folder_id,
+        )
+    )
+    cloud_item = result.scalar_one_or_none()
+    assert cloud_item is not None, (
+        "Create-folder success must upsert a CloudItem row before returning "
+        "(Plan 08 Task 2 behavior 1)"
+    )
+    assert cloud_item.kind == "folder", (
+        f"Upserted CloudItem must have kind='folder', got {cloud_item.kind!r}"
+    )
+    assert cloud_item.name == "My New Folder", (
+        f"Upserted CloudItem must have the correct name, got {cloud_item.name!r}"
+    )
+
+
+async def test_rename_success_updates_cloud_item_before_returning(async_client, db_session):
+    """PATCH rename success updates the renamed item in cloud_items before returning.
+
+    Plan 08 Task 2 behavior 1: Successful rename must route through centralized
+    reconciliation — upsert_cloud_item must be called with the new name so that
+    navigation shows the updated name immediately without a full provider re-list.
+
+    FAILS: Current rename route does not call upsert_cloud_item on success.
+    """
+    from storage.cloud_base import MUT_KIND_UPDATED, MUT_REASON_RENAMED
+    from db.models import CloudItem
+    from sqlalchemy import select
+
+    auth = await _create_user_and_token(db_session)
+    conn = await _create_cloud_connection(db_session, auth["user"].id)
+
+    # Pre-seed an existing CloudItem so the upsert path can update it
+    existing_provider_id = "rename_target_item"
+    test_parent_ref = "rename_parent_folder"
+    existing_item = CloudItem(
+        id=_uuid.uuid4(),
+        user_id=auth["user"].id,
+        connection_id=conn.id,
+        provider_item_id=existing_provider_id,
+        name="Old Report.pdf",
+        kind="file",
+        parent_ref=test_parent_ref,
+    )
+    db_session.add(existing_item)
+    await db_session.commit()
+
+    mock_adapter = _make_mock_mutable_adapter(
+        rename_result={
+            "kind": MUT_KIND_UPDATED,
+            "reason": MUT_REASON_RENAMED,
+            "provider_item_id": existing_provider_id,
+            "name": "New Report.pdf",
+        }
+    )
+    payload = {"new_name": "New Report.pdf", "etag": "v1"}
+
+    with patch("storage.cloud_backend_factory.build_mutable_cloud_adapter", return_value=mock_adapter):
+        resp = await async_client.patch(
+            f"/api/cloud/connections/{conn.id}/items/{existing_provider_id}/rename",
+            headers=auth["headers"],
+            json=payload,
+        )
+
+    assert resp.status_code == 200, f"Expected 200 for rename, got {resp.status_code}"
+    body = resp.json()
+    assert body.get("kind") == "renamed"
+
+    # After successful rename, CloudItem name must be updated
+    result = await db_session.execute(
+        select(CloudItem).where(
+            CloudItem.connection_id == conn.id,
+            CloudItem.provider_item_id == existing_provider_id,
+        )
+    )
+    cloud_item = result.scalar_one_or_none()
+    assert cloud_item is not None, "Existing CloudItem must still exist after rename"
+    assert cloud_item.name == "New Report.pdf", (
+        f"Renamed CloudItem must have the new name, got {cloud_item.name!r} "
+        f"(Plan 08 Task 2 behavior 1)"
+    )
+
+
+async def test_create_folder_success_marks_parent_folder_non_fresh(async_client, db_session):
+    """POST create-folder success marks the parent folder as non-fresh.
+
+    Plan 08 Task 2 behavior 1: After a successful create-folder, the parent folder
+    listing has changed — the folder state must be invalidated so the next browse
+    triggers a provider re-list.
+
+    FAILS: Current create-folder route does not update folder state on success.
+    """
+    from storage.cloud_base import MUT_KIND_FOLDER, MUT_REASON_CREATED
+    from db.models import CloudFolderState
+    from sqlalchemy import select
+    import uuid as _uuid2
+
+    auth = await _create_user_and_token(db_session)
+    conn = await _create_cloud_connection(db_session, auth["user"].id)
+
+    test_parent_ref = "folder_freshness_parent"
+    new_folder_id = str(_uuid2.uuid4())
+
+    mock_adapter = _make_mock_mutable_adapter(
+        create_folder_result={
+            "kind": MUT_KIND_FOLDER,
+            "reason": MUT_REASON_CREATED,
+            "provider_item_id": new_folder_id,
+            "name": "New Subfolder",
+            "parent_ref": test_parent_ref,
+        }
+    )
+
+    payload = {"parent_ref": test_parent_ref, "name": "New Subfolder"}
+
+    with patch("storage.cloud_backend_factory.build_mutable_cloud_adapter", return_value=mock_adapter):
+        resp = await async_client.post(
+            f"/api/cloud/connections/{conn.id}/folders",
+            headers=auth["headers"],
+            json=payload,
+        )
+
+    assert resp.status_code == 201
+    assert resp.json().get("kind") == "folder"
+
+    # Parent folder state must be invalidated
+    result = await db_session.execute(
+        select(CloudFolderState).where(
+            CloudFolderState.connection_id == conn.id,
+            CloudFolderState.parent_ref == test_parent_ref,
+        )
+    )
+    fs = result.scalar_one_or_none()
+    assert fs is not None, (
+        "Create-folder success must create/update a CloudFolderState row for the parent "
+        "(Plan 08 Task 2 behavior 1)"
+    )
+
+
+async def test_rename_success_marks_parent_folder_non_fresh(async_client, db_session):
+    """PATCH rename success marks the parent folder as non-fresh.
+
+    Plan 08 Task 2 behavior 1: After a successful rename, the parent folder listing
+    has changed — folder state must be invalidated so the next browse re-lists.
+
+    FAILS: Current rename route does not update folder state on success.
+    """
+    from storage.cloud_base import MUT_KIND_UPDATED, MUT_REASON_RENAMED
+    from db.models import CloudFolderState, CloudItem
+    from sqlalchemy import select
+
+    auth = await _create_user_and_token(db_session)
+    conn = await _create_cloud_connection(db_session, auth["user"].id)
+
+    rename_parent_ref = "rename_freshness_parent"
+    item_provider_id = "rename_fresh_target"
+
+    # Seed item
+    existing_item = CloudItem(
+        id=_uuid.uuid4(),
+        user_id=auth["user"].id,
+        connection_id=conn.id,
+        provider_item_id=item_provider_id,
+        name="Original.pdf",
+        kind="file",
+        parent_ref=rename_parent_ref,
+    )
+    db_session.add(existing_item)
+    await db_session.commit()
+
+    mock_adapter = _make_mock_mutable_adapter(
+        rename_result={
+            "kind": MUT_KIND_UPDATED,
+            "reason": MUT_REASON_RENAMED,
+            "provider_item_id": item_provider_id,
+            "name": "Renamed.pdf",
+        }
+    )
+    payload = {"new_name": "Renamed.pdf", "etag": "v1"}
+
+    with patch("storage.cloud_backend_factory.build_mutable_cloud_adapter", return_value=mock_adapter):
+        resp = await async_client.patch(
+            f"/api/cloud/connections/{conn.id}/items/{item_provider_id}/rename",
+            headers=auth["headers"],
+            json=payload,
+        )
+
+    assert resp.status_code == 200
+    assert resp.json().get("kind") == "renamed"
+
+    # Parent folder state must be invalidated after rename
+    result = await db_session.execute(
+        select(CloudFolderState).where(
+            CloudFolderState.connection_id == conn.id,
+            CloudFolderState.parent_ref == rename_parent_ref,
+        )
+    )
+    fs = result.scalar_one_or_none()
+    assert fs is not None, (
+        "Rename success must create/update a CloudFolderState row for the parent folder "
+        "(Plan 08 Task 2 behavior 1)"
+    )
+
+
+async def test_rename_failed_does_not_mutate_cloud_items(async_client, db_session):
+    """Failed rename (stale/offline) must not mutate CloudItem rows.
+
+    Plan 08 Task 2 behavior 3: Only authoritative rename success may mutate
+    metadata. A stale or offline rename result must not update the CloudItem name —
+    that would corrupt the navigation cache with a phantom rename.
+
+    FAILS: Not yet enforced — establishes invariant for regression coverage.
+    """
+    from storage.cloud_base import MUT_KIND_STALE, MUT_REASON_ITEM_CHANGED
+    from db.models import CloudItem
+    from sqlalchemy import select
+
+    auth = await _create_user_and_token(db_session)
+    conn = await _create_cloud_connection(db_session, auth["user"].id)
+
+    original_name = "Cannot Be Renamed.pdf"
+    item_provider_id = "immutable_item_id"
+    test_parent_ref = "immutable_parent"
+
+    existing_item = CloudItem(
+        id=_uuid.uuid4(),
+        user_id=auth["user"].id,
+        connection_id=conn.id,
+        provider_item_id=item_provider_id,
+        name=original_name,
+        kind="file",
+        parent_ref=test_parent_ref,
+    )
+    db_session.add(existing_item)
+    await db_session.commit()
+
+    mock_adapter = _make_mock_mutable_adapter(
+        rename_result={"kind": MUT_KIND_STALE, "reason": MUT_REASON_ITEM_CHANGED}
+    )
+    payload = {"new_name": "Different Name.pdf", "etag": "stale-v0"}
+
+    with patch("storage.cloud_backend_factory.build_mutable_cloud_adapter", return_value=mock_adapter):
+        resp = await async_client.patch(
+            f"/api/cloud/connections/{conn.id}/items/{item_provider_id}/rename",
+            headers=auth["headers"],
+            json=payload,
+        )
+
+    assert resp.status_code == 409  # stale returns 409
+
+    # CloudItem name must be unchanged after a stale rename
+    result = await db_session.execute(
+        select(CloudItem).where(
+            CloudItem.connection_id == conn.id,
+            CloudItem.provider_item_id == item_provider_id,
+        )
+    )
+    cloud_item = result.scalar_one_or_none()
+    assert cloud_item is not None
+    assert cloud_item.name == original_name, (
+        f"Failed rename must not change CloudItem name; "
+        f"expected {original_name!r}, got {cloud_item.name!r} "
+        f"(Plan 08 Task 2 behavior 3)"
+    )