kite/backend/tests/test_cloud_mutations.py

"""
Phase 13 Plan 01 — TDD RED: Endpoint and mutation-result contracts.

Covers D-02 through D-11 and D-18 typed result semantics:
  - Open/preview: authorized download fallback, binary-only preview (D-02, D-18)
  - Upload: conflict dialog semantics — kind/reason body, no silent overwrite (D-03, D-04)
  - Create folder: collision auto-naming, typed conflict result (D-05, D-06)
  - Rename / move: stale-metadata guard, typed conflict result (D-05, D-06, D-07)
  - Delete: trash vs permanent disclosure, nested-folder warning (D-09, D-10, D-11)
  - Move: same-connection restriction, invalid-destination rejection (D-08, D-09)
  - Security: IDOR, admin block, credential/bytes never in response

All tests FAIL against the current codebase because Phase 13 mutation routes
do not yet exist.  They define the contract Phase 13 implementation must satisfy.

Requirements: CONN-01, CONN-02, CONN-03, CLOUD-02, CLOUD-03, CLOUD-04, CLOUD-05,
              CLOUD-06, CLOUD-07, CLOUD-09
"""
from __future__ import annotations

import uuid as _uuid
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

pytestmark = pytest.mark.asyncio

from tests.conftest import _TEST_USER_AGENT


# ── Shared helpers ─────────────────────────────────────────────────────────────


async def _create_user_and_token(session, role: str = "user"):
    """Create User + Quota + JWT. Mirrors the pattern from test_cloud_security.py."""
    from db.models import User, Quota
    from services.auth import hash_password, create_access_token

    user_id = _uuid.uuid4()
    user = User(
        id=user_id,
        handle=f"mut_user_{user_id.hex[:8]}",
        email=f"mut_{user_id.hex[:8]}@example.com",
        password_hash=hash_password("Testpassword123!"),
        role=role,
        is_active=True,
        password_must_change=False,
    )
    quota = Quota(user_id=user_id, limit_bytes=104857600, used_bytes=0)
    session.add(user)
    session.add(quota)
    await session.commit()
    await session.refresh(user)

    token = create_access_token(str(user_id), role, user_agent=_TEST_USER_AGENT)
    return {
        "user": user,
        "token": token,
        "headers": {
            "Authorization": f"Bearer {token}",
            "User-Agent": _TEST_USER_AGENT,
        },
    }


async def _create_cloud_connection(
    session,
    user_id,
    provider: str = "google_drive",
    name: str = "My Drive",
    status: str = "ACTIVE",
):
    """Create a CloudConnection row for mutation test fixtures.

    Encrypts with the backend's actual settings key so that mutation endpoint
    credential decryption works in the test environment.
    """
    from db.models import CloudConnection
    from storage.cloud_utils import encrypt_credentials
    from config import settings

    master_key = settings.cloud_creds_key.encode()
    creds_enc = encrypt_credentials(
        master_key,
        str(user_id),
        {"access_token": "tok", "refresh_token": "ref"},
    )
    conn = CloudConnection(
        id=_uuid.uuid4(),
        user_id=user_id,
        provider=provider,
        display_name=name,
        credentials_enc=creds_enc,
        status=status,
    )
    session.add(conn)
    await session.commit()
    return conn


def _make_mock_mutable_adapter(
    upload_result: dict | None = None,
    create_folder_result: dict | None = None,
    rename_result: dict | None = None,
    move_result: dict | None = None,
    delete_result: dict | None = None,
):
    """Return a mock MutableCloudResourceAdapter for testing mutation routes.

    Default results produce the canonical success outcomes unless overridden.
    """
    from storage.cloud_base import (
        MUT_KIND_UPLOADED, MUT_KIND_FOLDER, MUT_KIND_UPDATED, MUT_KIND_DELETED,
        MUT_REASON_CREATED, MUT_REASON_RENAMED, MUT_REASON_MOVED, MUT_REASON_TRASHED,
    )
    import uuid as _uuid2

    adapter = AsyncMock()
    # get_object is used by preview/download routes
    adapter.get_object = AsyncMock(return_value=b"fake binary content")
    adapter._normalize_error = MagicMock(return_value={"kind": "error", "reason": "provider_error"})

    adapter.upload_file = AsyncMock(return_value=upload_result or {
        "kind": MUT_KIND_UPLOADED,
        "reason": MUT_REASON_CREATED,
        "provider_item_id": str(_uuid2.uuid4()),
        "name": "uploaded_file.pdf",
        "parent_ref": None,
        "size": 42,
    })
    adapter.create_folder = AsyncMock(return_value=create_folder_result or {
        "kind": MUT_KIND_FOLDER,
        "reason": MUT_REASON_CREATED,
        "provider_item_id": str(_uuid2.uuid4()),
        "name": "New Folder",
        "parent_ref": None,
    })
    adapter.rename = AsyncMock(return_value=rename_result or {
        "kind": MUT_KIND_UPDATED,
        "reason": MUT_REASON_RENAMED,
        "provider_item_id": "fake_item_id",
        "name": "Updated Report.pdf",
    })
    adapter.move = AsyncMock(return_value=move_result or {
        "kind": MUT_KIND_UPDATED,
        "reason": MUT_REASON_MOVED,
        "provider_item_id": "fake_item_id",
        "destination_parent_ref": "dest_folder_ref",
    })
    adapter.delete = AsyncMock(return_value=delete_result or {
        "kind": MUT_KIND_DELETED,
        "reason": MUT_REASON_TRASHED,
        "provider_item_id": "fake_file_id",
    })
    return adapter


# ── T-13-01: Open endpoint — D-02 authorized download fallback ─────────────────


async def test_open_file_returns_authorized_download_url(async_client, db_session):
    """GET /api/cloud/connections/{id}/items/{item_id}/open returns an authorized URL.

    D-02: Provider credentials and raw provider URLs must never appear in the response.
    The endpoint must serve an authorized DocuVault-scoped download URL only.

    FAILS: Phase 13 route does not exist yet.
    """
    auth = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, auth["user"].id)

    resp = await async_client.get(
        f"/api/cloud/connections/{conn.id}/items/fake_item_id/open",
        headers=auth["headers"],
    )
    # Phase 13 route not yet implemented — expect 404 (method-not-found)
    assert resp.status_code == 200, (
        f"Expected 200 with authorized URL body, got {resp.status_code}: {resp.text}"
    )
    body = resp.json()
    assert "url" in body, "Response must include authorized download URL"
    # Provider URL must never be exposed
    for forbidden in ("access_token", "refresh_token", "credentials_enc", "client_secret"):
        assert forbidden not in resp.text, (
            f"Response must not expose '{forbidden}' (T-13-02)"
        )


async def test_open_file_foreign_user_blocked(async_client, db_session):
    """User2 cannot open a file on User1's connection — IDOR protection (T-13-01).

    FAILS: Phase 13 route does not exist yet.
    """
    auth1 = await _create_user_and_token(db_session)
    auth2 = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, auth1["user"].id)

    resp = await async_client.get(
        f"/api/cloud/connections/{conn.id}/items/fake_item_id/open",
        headers=auth2["headers"],
    )
    assert resp.status_code == 404, (
        f"Expected 404 IDOR block, got {resp.status_code}"
    )


async def test_open_file_admin_blocked(async_client, db_session):
    """Admin token cannot access open endpoint — admin blocked from cloud content (T-13-01).

    FAILS: Phase 13 route does not exist yet.
    """
    auth_admin = await _create_user_and_token(db_session, role="admin")
    auth_user = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, auth_user["user"].id)

    resp = await async_client.get(
        f"/api/cloud/connections/{conn.id}/items/fake_item_id/open",
        headers=auth_admin["headers"],
    )
    assert resp.status_code in (403, 404), (
        f"Expected admin block (403/404), got {resp.status_code}"
    )


# ── T-13-02: Preview endpoint — D-18 binary-only, no device download ──────────


async def test_preview_binary_file_returns_content(async_client, db_session):
    """GET /api/cloud/connections/{id}/items/{item_id}/preview returns binary content.

    D-18: Only supported binary file formats are previewed. Provider credentials
    and raw URLs must never be in the response. Content streams inline — not as
    a browser-to-device file download.

    FAILS: Phase 13 route does not exist yet.
    """
    auth = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, auth["user"].id)

    resp = await async_client.get(
        f"/api/cloud/connections/{conn.id}/items/fake_item_id/preview",
        headers=auth["headers"],
    )
    assert resp.status_code == 200, (
        f"Expected 200 with preview content, got {resp.status_code}"
    )
    # Must not trigger a device download
    cd = resp.headers.get("content-disposition", "")
    assert "attachment" not in cd, (
        "Preview must not produce a Content-Disposition: attachment header (D-18)"
    )


async def test_preview_unsupported_format_returns_typed_error(async_client, db_session):
    """Preview of an unsupported format returns typed kind='unsupported_preview' body.

    D-18 / D-02: Unsupported formats fall back to authorized download. The
    response must carry a typed {kind, reason} body so the frontend knows to
    route to the authorized download flow rather than infer from a raw error.

    FAILS: Phase 13 route does not exist yet.
    """
    auth = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, auth["user"].id)

    resp = await async_client.get(
        f"/api/cloud/connections/{conn.id}/items/fake_docx_item/preview",
        headers=auth["headers"],
    )
    assert resp.status_code in (200, 409, 422), (
        f"Expected typed response for unsupported preview, got {resp.status_code}"
    )
    body = resp.json()
    assert "kind" in body, "Must include 'kind' field in unsupported preview response"
    assert body["kind"] == "unsupported_preview", (
        f"Expected kind='unsupported_preview', got {body.get('kind')!r}"
    )
    assert "reason" in body, "Must include 'reason' field"


# ── Upload conflict — D-03, D-04 typed kind/reason body ─────────────────────


async def test_upload_same_name_returns_conflict_kind(async_client, db_session):
    """POST /api/cloud/connections/{id}/items/upload with same-name file returns typed conflict.

    D-03: Same-name upload must NEVER overwrite silently. The response body must
    carry {kind: 'conflict', reason: 'name_collision'} so the frontend can show
    the conflict resolution dialog.

    The mock adapter returns a conflict result to simulate same-name detection.
    """
    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)

    # Seed a conflicting cloud item so the fast-path conflict check triggers
    from db.models import CloudItem
    existing = CloudItem(
        id=_uuid.uuid4(),
        user_id=auth["user"].id,
        connection_id=conn.id,
        provider_item_id="existing_report_id",
        name="report.pdf",
        kind="file",
        parent_ref="root",
    )
    db_session.add(existing)
    await db_session.commit()

    files = {"file": ("report.pdf", b"%PDF-1.4 fake", "application/pdf")}
    data = {"parent_ref": "root", "filename": "report.pdf"}

    resp = await async_client.post(
        f"/api/cloud/connections/{conn.id}/items/upload",
        headers=auth["headers"],
        files=files,
        data=data,
    )
    assert resp.status_code in (200, 409), (
        f"Expected 200 or 409 for upload conflict signal, got {resp.status_code}"
    )
    body = resp.json()
    assert "kind" in body, "Conflict response must include 'kind' field (D-03)"
    assert body["kind"] == "conflict", f"Expected kind='conflict', got {body.get('kind')!r}"
    assert "reason" in body, "Conflict response must include 'reason' field"
    assert body["reason"] == "name_collision", (
        f"Expected reason='name_collision', got {body.get('reason')!r}"
    )


async def test_upload_response_excludes_credentials(async_client, db_session):
    """Upload response must never contain access_token, refresh_token, or credentials_enc.

    T-13-02: Credential secrecy invariant must hold on every upload response shape.
    Mock adapter used so the test verifies the route shape without real provider I/O.
    """
    auth = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, auth["user"].id)

    mock_adapter = _make_mock_mutable_adapter()
    files = {"file": ("doc.txt", b"hello world", "text/plain")}
    data = {"parent_ref": "root", "filename": "doc.txt"}

    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}/items/upload",
            headers=auth["headers"],
            files=files,
            data=data,
        )
    for forbidden in ("access_token", "refresh_token", "credentials_enc", "client_secret"):
        assert forbidden not in resp.text, (
            f"Upload response must not expose '{forbidden}' (T-13-02)"
        )


# ── Create folder — D-05 collision naming, typed conflict ───────────────────


async def test_create_folder_returns_typed_result(async_client, db_session):
    """POST /api/cloud/connections/{id}/folders creates a folder and returns typed result.

    D-05: Collision auto-name produces 'Projects (1)' not an error.
    The success response must include the item's kind and provider_item_id.
    Mock adapter used so the test verifies the route shape without real provider I/O.
    """
    auth = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, auth["user"].id)

    mock_adapter = _make_mock_mutable_adapter()
    payload = {"parent_ref": None, "name": "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 Created for new folder, got {resp.status_code}: {resp.text}"
    )
    body = resp.json()
    assert "provider_item_id" in body, "Response must include provider_item_id"
    assert "kind" in body and body["kind"] == "folder", (
        "Response must include kind='folder'"
    )


async def test_create_folder_collision_returns_auto_name(async_client, db_session):
    """POST create-folder with a colliding name returns result with counter-suffixed name.

    D-05: Automatic non-conflicting name: 'Projects (1)', 'Projects (2)', etc.
    Must not return an error — the backend auto-resolves naming collision.
    Mock adapter used so the test verifies the route shape without real provider I/O.
    """
    from storage.cloud_base import MUT_KIND_FOLDER, MUT_REASON_CREATED

    auth = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, auth["user"].id)

    # Adapter auto-renames on collision — returns the suffixed name
    import uuid as _uuid2
    mock_adapter = _make_mock_mutable_adapter(
        create_folder_result={
            "kind": MUT_KIND_FOLDER,
            "reason": MUT_REASON_CREATED,
            "provider_item_id": str(_uuid2.uuid4()),
            "name": "Projects (1)",
            "parent_ref": None,
        }
    )
    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 with auto-named folder, got {resp.status_code}"
    )
    body = resp.json()
    assert "name" in body, "Response must include resolved name"


# ── Rename — D-05, D-07 stale-metadata guard ────────────────────────────────


async def test_rename_item_returns_typed_result(async_client, db_session):
    """PATCH /api/cloud/connections/{id}/items/{item_id}/rename with valid new name.

    Success body must include kind and updated name — no raw provider errors.
    Mock adapter used so the test verifies the route shape without real provider I/O.
    """
    auth = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, auth["user"].id)

    mock_adapter = _make_mock_mutable_adapter()
    payload = {"new_name": "Updated 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 == 200, (
        f"Expected 200 for rename, got {resp.status_code}: {resp.text}"
    )
    body = resp.json()
    assert "kind" in body, "Rename response must include 'kind' field"


async def test_rename_stale_etag_returns_stale_kind(async_client, db_session):
    """Rename with a stale etag returns typed kind='stale' body (D-07).

    The backend must detect externally-changed metadata, stop the mutation,
    return {kind: 'stale', reason: 'item_changed'}, and require a retry
    after the user acknowledges the listing refresh.
    Mock adapter simulates stale-etag detection.
    """
    from storage.cloud_base import MUT_KIND_STALE, MUT_REASON_ITEM_CHANGED

    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_STALE, "reason": MUT_REASON_ITEM_CHANGED}
    )
    payload = {"new_name": "Report.pdf", "etag": "stale-etag-abc"}

    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 == 409, (
        f"Expected 409 Conflict for stale rename, got {resp.status_code}"
    )
    body = resp.json()
    assert "kind" in body and body["kind"] == "stale", (
        f"Expected kind='stale', got {body.get('kind')!r}"
    )
    assert "reason" in body and body["reason"] == "item_changed", (
        f"Expected reason='item_changed', got {body.get('reason')!r}"
    )


async def test_rename_foreign_user_blocked(async_client, db_session):
    """User2 cannot rename an item on User1's connection — IDOR (T-13-01).

    FAILS: Phase 13 route does not exist yet.
    """
    auth1 = await _create_user_and_token(db_session)
    auth2 = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, auth1["user"].id)

    payload = {"new_name": "Hacked.pdf", "etag": "v1"}
    resp = await async_client.patch(
        f"/api/cloud/connections/{conn.id}/items/fake_item_id/rename",
        headers=auth2["headers"],
        json=payload,
    )
    assert resp.status_code == 404, (
        f"Expected 404 IDOR block, got {resp.status_code}"
    )


# ── Move — D-08, D-09 same-connection, invalid-destination ──────────────────


async def test_move_item_same_connection_succeeds(async_client, db_session):
    """POST /api/cloud/connections/{id}/items/{item_id}/move within same connection.

    D-08: Moves are restricted to items within the same cloud connection.
    Success returns typed body with updated parent_ref.
    Mock adapter used so the test verifies the route shape without real provider I/O.
    """
    auth = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, auth["user"].id)

    mock_adapter = _make_mock_mutable_adapter()
    payload = {
        "destination_parent_ref": "dest_folder_ref",
        "etag": "v1",
    }

    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}/items/fake_item_id/move",
            headers=auth["headers"],
            json=payload,
        )
    assert resp.status_code == 200, (
        f"Expected 200 for move, got {resp.status_code}: {resp.text}"
    )
    body = resp.json()
    assert "kind" in body, "Move response must include 'kind' field"


async def test_move_item_self_as_destination_rejected(async_client, db_session):
    """Moving a folder into itself returns typed kind='invalid_destination' (D-09).

    Invalid destinations (self, descendants) must be rejected by the backend
    even if the frontend pre-screens them.

    FAILS: Phase 13 route does not exist yet.
    """
    auth = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, auth["user"].id)

    payload = {
        "destination_parent_ref": "fake_item_id",  # same as source
        "etag": "v1",
    }
    resp = await async_client.post(
        f"/api/cloud/connections/{conn.id}/items/fake_item_id/move",
        headers=auth["headers"],
        json=payload,
    )
    assert resp.status_code in (400, 409, 422), (
        f"Expected rejection for self-destination move, got {resp.status_code}"
    )
    body = resp.json()
    assert "kind" in body and body["kind"] == "invalid_destination", (
        f"Expected kind='invalid_destination', got {body.get('kind')!r}"
    )


async def test_move_item_cross_connection_rejected(async_client, db_session):
    """Move across different connections returns typed kind='invalid_destination' (D-08).

    Cross-provider transfer is explicitly out of scope. The backend must reject
    any destination_connection_id that does not match the source connection.

    FAILS: Phase 13 route does not exist yet.
    """
    auth = await _create_user_and_token(db_session)
    conn1 = await _create_cloud_connection(db_session, auth["user"].id, name="Drive 1")
    conn2 = await _create_cloud_connection(db_session, auth["user"].id, name="Drive 2")

    payload = {
        "destination_parent_ref": "some_folder_in_conn2",
        "destination_connection_id": str(conn2.id),  # different connection
        "etag": "v1",
    }
    resp = await async_client.post(
        f"/api/cloud/connections/{conn1.id}/items/fake_item_id/move",
        headers=auth["headers"],
        json=payload,
    )
    assert resp.status_code in (400, 409, 422), (
        f"Expected rejection for cross-connection move, got {resp.status_code}"
    )
    body = resp.json()
    assert "kind" in body and body["kind"] == "invalid_destination", (
        f"Expected kind='invalid_destination', got {body.get('kind')!r}"
    )


# ── Delete — D-10, D-11 confirmation, trash vs permanent ────────────────────


async def test_delete_file_returns_typed_result(async_client, db_session):
    """DELETE /api/cloud/connections/{id}/items/{item_id} returns typed result.

    D-11: Response must indicate whether trash or permanent delete was performed
    via {kind: 'deleted', reason: 'trashed'} or {kind: 'deleted', reason: 'permanent'}.
    Mock adapter used so the test verifies the route shape without real provider I/O.
    """
    auth = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, auth["user"].id)

    mock_adapter = _make_mock_mutable_adapter()

    with patch("storage.cloud_backend_factory.build_mutable_cloud_adapter", return_value=mock_adapter):
        resp = await async_client.delete(
            f"/api/cloud/connections/{conn.id}/items/fake_file_id",
            headers=auth["headers"],
        )
    assert resp.status_code in (200, 204), (
        f"Expected 200/204 for delete, got {resp.status_code}: {resp.text}"
    )
    if resp.status_code == 200:
        body = resp.json()
        assert "kind" in body and body["kind"] == "deleted", (
            f"Expected kind='deleted', got {body.get('kind')!r}"
        )
        assert "reason" in body and body["reason"] in ("trashed", "permanent"), (
            f"Expected reason 'trashed' or 'permanent', got {body.get('reason')!r}"
        )


async def test_delete_foreign_user_blocked(async_client, db_session):
    """User2 cannot delete items on User1's connection (T-13-01).

    FAILS: Phase 13 route does not exist yet.
    """
    auth1 = await _create_user_and_token(db_session)
    auth2 = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, auth1["user"].id)

    resp = await async_client.delete(
        f"/api/cloud/connections/{conn.id}/items/fake_file_id",
        headers=auth2["headers"],
    )
    assert resp.status_code == 404, (
        f"Expected 404 IDOR block, got {resp.status_code}"
    )


async def test_delete_response_excludes_provider_urls_and_tokens(async_client, db_session):
    """Delete response must not expose provider URLs, tokens, or credentials_enc (T-13-02).

    Mock adapter used so the test verifies the credential-secrecy invariant
    without real provider I/O.
    """
    auth = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, auth["user"].id)

    mock_adapter = _make_mock_mutable_adapter()

    with patch("storage.cloud_backend_factory.build_mutable_cloud_adapter", return_value=mock_adapter):
        resp = await async_client.delete(
            f"/api/cloud/connections/{conn.id}/items/fake_file_id",
            headers=auth["headers"],
        )
    for forbidden in ("access_token", "refresh_token", "credentials_enc", "client_secret"):
        assert forbidden not in resp.text, (
            f"Delete response must not expose '{forbidden}' (T-13-02)"
        )


# ── Typed conflict/error kinds — T-13-03 ─────────────────────────────────────


async def test_mutation_offline_connection_returns_offline_kind(async_client, db_session):
    """Any mutation on an offline connection returns a typed non-500 body.

    D-15: Transient provider unreachability must not destroy data. The response
    must carry a typed kind body so the frontend can show actionable retry UI.
    Mock adapter simulates offline/unreachable provider.
    """
    from storage.cloud_base import MUT_KIND_OFFLINE, MUT_REASON_PROVIDER_OFFLINE

    auth = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, auth["user"].id, status="ACTIVE")

    mock_adapter = _make_mock_mutable_adapter(
        rename_result={"kind": MUT_KIND_OFFLINE, "reason": MUT_REASON_PROVIDER_OFFLINE}
    )
    payload = {"new_name": "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,
        )
    # Offline provider must return typed error body, not 500
    assert resp.status_code != 500, (
        "Offline provider must not return a 500 — use typed kind='offline' body"
    )


async def test_mutation_reauth_required_returns_reauth_kind(async_client, db_session):
    """Mutation with expired credentials returns typed kind='reauth_required' body.

    D-13: Credential-related failures trigger automatic health re-evaluation.
    The response must carry {kind: 'reauth_required', reason: 'token_expired'}.
    Uses an AUTH_FAILED connection that triggers the reauth response path.
    """
    auth = await _create_user_and_token(db_session)
    # AUTH_FAILED connections return reauth_required when rename is attempted.
    # The mock adapter simulates the reauth result.
    conn = await _create_cloud_connection(db_session, auth["user"].id, status="AUTH_FAILED")

    from storage.cloud_base import MUT_KIND_REAUTH, MUT_REASON_TOKEN_EXPIRED

    mock_adapter = _make_mock_mutable_adapter(
        rename_result={"kind": MUT_KIND_REAUTH, "reason": MUT_REASON_TOKEN_EXPIRED}
    )
    payload = {"new_name": "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, 401, 409), (
        f"Expected typed reauth response, got {resp.status_code}"
    )
    if resp.status_code != 200:
        body = resp.json()
        assert "kind" in body and body["kind"] == "reauth_required", (
            f"Expected kind='reauth_required', got {body.get('kind')!r}"
        )


# ── Unsupported operation — D-18 ─────────────────────────────────────────────


async def test_unsupported_operation_returns_typed_kind(async_client, db_session):
    """Operation not supported by provider returns typed kind='unsupported_operation'.

    Providers that cannot honor a mutation (e.g. read-only WebDAV) must return
    {kind: 'unsupported_operation', reason: 'provider_unsupported'} rather than
    a 500 error or silent pass.
    Mock adapter simulates an unsupported provider.
    """
    from storage.cloud_base import MUT_KIND_UNSUPPORTED, MUT_REASON_NOT_SUPPORTED

    auth = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, auth["user"].id, provider="webdav")

    mock_adapter = _make_mock_mutable_adapter(
        create_folder_result={"kind": MUT_KIND_UNSUPPORTED, "reason": MUT_REASON_NOT_SUPPORTED}
    )
    payload = {"parent_ref": None, "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, (
        "Unsupported operation must return typed kind body, not 500 (D-18)"
    )
    if resp.status_code in (200, 201, 409, 422):
        body = resp.json()
        if "kind" in body:
            assert body["kind"] in ("folder", "unsupported_operation"), (
                f"Unexpected kind value: {body.get('kind')!r}"
            )


# ── Phase 13 Plan 05 Task 2: Upload route typed mechanics ─────────────────────


async def test_upload_success_returns_typed_uploaded_body(async_client, db_session):
    """POST upload with no conflict returns {kind: 'uploaded', provider_item_id, name, size}.

    Task 2 behavior 1: The upload route must return a stable typed 'uploaded' body so
    the queue can advance to the next file without Vue-side inference.
    Mock adapter returns the canonical uploaded result.
    """
    from storage.cloud_base import MUT_KIND_UPLOADED, MUT_REASON_CREATED

    auth = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, auth["user"].id)

    import uuid as _uuid2
    mock_adapter = _make_mock_mutable_adapter(
        upload_result={
            "kind": MUT_KIND_UPLOADED,
            "reason": MUT_REASON_CREATED,
            "provider_item_id": str(_uuid2.uuid4()),
            "name": "report.pdf",
            "parent_ref": "root",
            "size": 1024,
        }
    )

    files = {"file": ("report.pdf", b"%PDF-1.4 test", "application/pdf")}
    data = {"parent_ref": "root_no_conflict", "filename": "report.pdf"}

    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}/items/upload",
            headers=auth["headers"],
            files=files,
            data=data,
        )

    assert resp.status_code == 200, (
        f"Expected 200 for upload success, got {resp.status_code}: {resp.text}"
    )
    body = resp.json()
    assert body.get("kind") == "uploaded", (
        f"Expected kind='uploaded', got {body.get('kind')!r}"
    )
    assert "provider_item_id" in body, "Response must include provider_item_id"
    assert "name" in body, "Response must include name"
    assert "size" in body, "Response must include size"


async def test_upload_provider_retryable_error_returns_offline_kind(async_client, db_session):
    """POST upload on provider 503 returns typed {kind: 'offline', reason: 'provider_offline'}.

    Task 2 behavior 1: A transient provider error must pause the upload queue
    (D-04). The route must expose the typed 'offline' result so Vue can present
    the 'Retry / Skip / Cancel all' dialog — never a bare 500.
    """
    from storage.cloud_base import MUT_KIND_OFFLINE, MUT_REASON_PROVIDER_OFFLINE

    auth = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, auth["user"].id)

    mock_adapter = _make_mock_mutable_adapter(
        upload_result={
            "kind": MUT_KIND_OFFLINE,
            "reason": MUT_REASON_PROVIDER_OFFLINE,
        }
    )

    files = {"file": ("report.pdf", b"data", "application/pdf")}
    data = {"parent_ref": "root", "filename": "retryable_file.pdf"}

    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}/items/upload",
            headers=auth["headers"],
            files=files,
            data=data,
        )

    # Route must not raise a bare 500 — must return typed error
    assert resp.status_code != 500, (
        f"Retryable error must not produce a 500 (D-04), got {resp.status_code}: {resp.text}"
    )
    body = resp.json()
    assert "kind" in body, "Retryable error response must include 'kind' field (D-04)"
    # kind must be a recognized non-success result
    assert body.get("kind") in ("offline", "error", "reauth_required"), (
        f"Expected typed error kind, got {body.get('kind')!r}"
    )


async def test_upload_provider_conflict_returns_conflict_kind(async_client, db_session):
    """POST upload when provider detects conflict returns typed {kind: 'conflict'} body.

    Task 2 behavior 1: If the fast-path cache check misses but the provider
    returns a conflict (e.g. OneDrive 409 on createUploadSession), the route must
    surface the typed conflict body — not a raw 409 error — so the queue can show
    the conflict dialog.
    """
    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(
        upload_result={
            "kind": MUT_KIND_CONFLICT,
            "reason": MUT_REASON_NAME_COLLISION,
            "existing_name": "document.pdf",
        }
    )

    # No cached CloudItem — so fast path won't block
    files = {"file": ("document.pdf", b"data", "application/pdf")}
    data = {"parent_ref": "folder_no_cache", "filename": "document.pdf"}

    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}/items/upload",
            headers=auth["headers"],
            files=files,
            data=data,
        )

    assert resp.status_code in (200, 409), (
        f"Conflict must produce 200 or 409, not {resp.status_code}"
    )
    body = resp.json()
    assert body.get("kind") == "conflict", (
        f"Expected kind='conflict', got {body.get('kind')!r}"
    )
    assert body.get("reason") == "name_collision", (
        f"Expected reason='name_collision', got {body.get('reason')!r}"
    )


async def test_upload_reauth_result_is_typed_and_not_500(async_client, db_session):
    """POST upload when provider requires reauth returns typed {kind: 'reauth_required'}.

    Task 2 behavior 2: CONN-02 — refreshed credentials surface upward. The route
    must return typed reauth body so the queue knows to prompt reconnect.
    """
    from storage.cloud_base import MUT_KIND_REAUTH, MUT_REASON_TOKEN_EXPIRED

    auth = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, auth["user"].id)

    mock_adapter = _make_mock_mutable_adapter(
        upload_result={
            "kind": MUT_KIND_REAUTH,
            "reason": MUT_REASON_TOKEN_EXPIRED,
        }
    )

    files = {"file": ("report.pdf", b"data", "application/pdf")}
    data = {"parent_ref": "root", "filename": "auth_report.pdf"}

    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}/items/upload",
            headers=auth["headers"],
            files=files,
            data=data,
        )

    assert resp.status_code != 500, (
        f"Reauth result must not produce 500, got {resp.status_code}"
    )
    body = resp.json()
    assert "kind" in body, "Reauth response must include 'kind'"
    # Route may map reauth to a 4xx with typed body
    assert body.get("kind") in ("reauth_required", "error"), (
        f"Unexpected kind for reauth: {body.get('kind')!r}"
    )


async def test_upload_queue_control_semantics_are_backend_authored(async_client, db_session):
    """Upload response body contains all fields the queue needs to decide next action.

    Task 2 behavior 3: The queue-control fields (kind, reason, existing_name) must
    come from the backend response. Vue must never infer queue state from HTTP status
    alone — the body must have enough typed data.

    Success: body has kind, provider_item_id, name, size.
    Conflict: body has kind, reason, existing_name.
    """
    from storage.cloud_base import MUT_KIND_UPLOADED, MUT_REASON_CREATED

    auth = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, auth["user"].id)

    import uuid as _uuid2
    mock_adapter = _make_mock_mutable_adapter(
        upload_result={
            "kind": MUT_KIND_UPLOADED,
            "reason": MUT_REASON_CREATED,
            "provider_item_id": str(_uuid2.uuid4()),
            "name": "presentation.pdf",
            "parent_ref": "root",
            "size": 512,
        }
    )

    files = {"file": ("presentation.pdf", b"pptx", "application/pdf")}
    data = {"parent_ref": "root_unique_2", "filename": "presentation.pdf"}

    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}/items/upload",
            headers=auth["headers"],
            files=files,
            data=data,
        )

    assert resp.status_code == 200
    body = resp.json()

    # All queue-control fields must be present in success response
    required_fields = {"kind", "provider_item_id", "name", "size"}
    missing = required_fields - set(body.keys())
    assert not missing, (
        f"Upload success body missing queue-control fields: {missing} (Task 2 behavior 3)"
    )
    assert body["kind"] == "uploaded", f"Expected kind='uploaded', got {body['kind']!r}"


async def test_upload_keep_both_name_format(async_client, db_session):
    """keep_both_name() produces counter-before-extension format for collision resolution.

    Task 2: The service utility must be accessible from the route context for
    queue-side rename. The format must match 'Report (1).pdf' not 'Report.pdf (1)'.
    """
    from services.cloud_operations import keep_both_name

    # Verify the canonical keep-both naming format used by the upload queue
    assert keep_both_name("Report.pdf", 1) == "Report (1).pdf", (
        "keep_both_name must insert counter BEFORE extension (D-03)"
    )
    assert keep_both_name("archive.tar.gz", 1) == "archive (1).tar.gz", (
        "keep_both_name must handle compound extensions"
    )
    assert keep_both_name("Report.pdf", 3) == "Report (3).pdf", (
        "keep_both_name must use the passed counter value"
    )


async def test_upload_foreign_user_blocked(async_client, db_session):
    """User2 cannot upload to User1's connection — IDOR protection (T-13-01)."""
    auth1 = await _create_user_and_token(db_session)
    auth2 = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, auth1["user"].id)

    files = {"file": ("test.pdf", b"data", "application/pdf")}
    data = {"parent_ref": "root"}

    resp = await async_client.post(
        f"/api/cloud/connections/{conn.id}/items/upload",
        headers=auth2["headers"],
        files=files,
        data=data,
    )
    assert resp.status_code == 404, (
        f"Expected 404 IDOR block for foreign upload, got {resp.status_code}"
    )


# ── Phase 13 Plan 06 Task 1 (RED): Upload success routes through reconciliation ─


async def test_upload_success_upserts_cloud_item_before_returning(async_client, db_session):
    """POST upload success upserts the uploaded item into cloud_items before returning.

    Plan 06 Task 1 behavior 1: Successful upload must update navigation metadata
    through centralized reconciliation (cloud_items.upsert_cloud_item) before the
    route returns success — not as a fire-and-forget side effect.

    The returned 'uploaded' body's provider_item_id must have a matching CloudItem
    row in the DB by the time the response is received.

    FAILS: Upload route does not call upsert_cloud_item on success yet.
    """
    from storage.cloud_base import MUT_KIND_UPLOADED, MUT_REASON_CREATED
    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)

    new_provider_item_id = str(_uuid.uuid4())

    import uuid as _uuid2
    mock_adapter = _make_mock_mutable_adapter(
        upload_result={
            "kind": MUT_KIND_UPLOADED,
            "reason": MUT_REASON_CREATED,
            "provider_item_id": new_provider_item_id,
            "name": "reconcile_test.pdf",
            "parent_ref": "reconcile_folder",
            "size": 2048,
        }
    )

    files = {"file": ("reconcile_test.pdf", b"%PDF-1.4 reconcile", "application/pdf")}
    data = {"parent_ref": "reconcile_folder_unique", "filename": "reconcile_test.pdf"}

    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}/items/upload",
            headers=auth["headers"],
            files=files,
            data=data,
        )

    assert resp.status_code == 200, f"Expected 200 for upload success, got {resp.status_code}"
    body = resp.json()
    assert body.get("kind") == "uploaded"

    # After successful upload, 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_provider_item_id,
        )
    )
    cloud_item = result.scalar_one_or_none()
    assert cloud_item is not None, (
        "Upload success must upsert a CloudItem row before returning (Plan 06 Task 1 behavior 1)"
    )
    assert cloud_item.name == "reconcile_test.pdf", (
        f"Upserted CloudItem must have the correct name, got {cloud_item.name!r}"
    )
    assert cloud_item.kind == "file", (
        f"Upserted CloudItem must have kind='file', got {cloud_item.kind!r}"
    )


async def test_upload_success_marks_folder_freshness_stale(async_client, db_session):
    """POST upload success marks the parent folder freshness as stale (not fresh).

    Plan 06 Task 1 behavior 2: After a successful upload, the parent folder listing
    has changed — the folder state must be invalidated (set to 'warning' or 'stale')
    so the next browse triggers a provider re-list. Setting it 'fresh' with an old
    listing would be a lie.

    The implementation must NOT call apply_listing_and_finalize (which requires a full
    provider listing). It must call update_folder_state with a non-'fresh' state that
    signals the folder needs refreshing.

    FAILS: Upload route does not update folder freshness state yet.
    """
    from storage.cloud_base import MUT_KIND_UPLOADED, MUT_REASON_CREATED
    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)

    new_provider_item_id = str(_uuid.uuid4())
    test_parent_ref = "freshness_test_folder"

    mock_adapter = _make_mock_mutable_adapter(
        upload_result={
            "kind": MUT_KIND_UPLOADED,
            "reason": MUT_REASON_CREATED,
            "provider_item_id": new_provider_item_id,
            "name": "freshness_test.pdf",
            "parent_ref": test_parent_ref,
            "size": 512,
        }
    )

    files = {"file": ("freshness_test.pdf", b"content", "application/pdf")}
    data = {"parent_ref": test_parent_ref, "filename": "freshness_test.pdf"}

    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}/items/upload",
            headers=auth["headers"],
            files=files,
            data=data,
        )

    assert resp.status_code == 200
    assert resp.json().get("kind") == "uploaded"

    # Parent folder state must be refreshed/invalidated — not left as stale from before
    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()
    # After upload, a folder state row must exist to signal refresh needed
    assert fs is not None, (
        "Upload success must create/update a CloudFolderState row for the parent folder "
        "(Plan 06 Task 1 behavior 2)"
    )


async def test_upload_failed_does_not_mutate_cloud_items(async_client, db_session):
    """Failed, skipped, and canceled queue decisions do not create CloudItem rows.

    Plan 06 Task 1 behavior 3: Only authoritative upload success may mutate listing
    state. An 'offline' or 'reauth_required' upload result must not upsert a CloudItem
    row — that would corrupt the navigation cache with phantom items.

    FAILS: Current behavior is correct (no reconcile call at all), but this test
    establishes the invariant explicitly for regression.
    """
    from storage.cloud_base import MUT_KIND_OFFLINE, MUT_REASON_PROVIDER_OFFLINE
    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)

    before_count_result = await db_session.execute(
        select(CloudItem).where(CloudItem.connection_id == conn.id)
    )
    before_count = len(before_count_result.scalars().all())

    mock_adapter = _make_mock_mutable_adapter(
        upload_result={"kind": MUT_KIND_OFFLINE, "reason": MUT_REASON_PROVIDER_OFFLINE}
    )

    files = {"file": ("offline_test.pdf", b"data", "application/pdf")}
    data = {"parent_ref": "some_folder", "filename": "offline_phantom.pdf"}

    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}/items/upload",
            headers=auth["headers"],
            files=files,
            data=data,
        )

    assert resp.status_code != 500

    # No new CloudItem must have been created for the failed upload
    after_result = await db_session.execute(
        select(CloudItem).where(CloudItem.connection_id == conn.id)
    )
    after_count = len(after_result.scalars().all())
    assert after_count == before_count, (
        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)"
    )


# ── Phase 13 Plan 09 Task 1 (RED): Move — stale guard, descendant safety, reconciliation ─


async def test_move_stale_etag_returns_stale_kind_and_refreshes_folder(async_client, db_session):
    """POST move with stale etag returns kind='stale' and refreshes parent folder state (D-07).

    Plan 09 Task 1 behavior 2: Stale etag mismatch must stop the move, mark the source
    folder state as non-fresh, and return typed {kind: 'stale', reason: 'item_changed'}
    so the frontend knows to re-list before retrying.

    FAILS: Current move route does not handle stale results from the provider.
    """
    from storage.cloud_base import MUT_KIND_STALE, MUT_REASON_ITEM_CHANGED
    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)

    # Pre-seed a CloudItem so we have a known parent_ref
    test_parent_ref = "move_stale_source_folder"
    existing_item = CloudItem(
        id=_uuid.uuid4(),
        user_id=auth["user"].id,
        connection_id=conn.id,
        provider_item_id="move_stale_item_ref",
        name="Report.pdf",
        kind="file",
        parent_ref=test_parent_ref,
    )
    db_session.add(existing_item)
    await db_session.commit()

    mock_adapter = _make_mock_mutable_adapter(
        move_result={"kind": MUT_KIND_STALE, "reason": MUT_REASON_ITEM_CHANGED}
    )
    payload = {"destination_parent_ref": "dest_folder", "etag": "stale-v0"}

    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}/items/move_stale_item_ref/move",
            headers=auth["headers"],
            json=payload,
        )

    # Must return typed stale body at HTTP 409
    assert resp.status_code == 409, (
        f"Expected 409 for stale move, 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}"
    )

    # Source 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 move must create/update a CloudFolderState row for the source folder "
        "(Plan 09 Task 1 behavior 2 — folder refresh required before retry)"
    )
    assert fs.refresh_state != "fresh", (
        f"Stale move must mark source folder as non-fresh; got refresh_state={fs.refresh_state!r}"
    )


async def test_move_descendant_destination_rejected(async_client, db_session):
    """POST move rejects descendant folder as destination (D-09) — backend validates independently.

    Plan 09 Task 1 behavior 1: The backend must reject a move where the destination
    folder is a descendant of the item being moved, even if the frontend pre-screened it.
    The provider must NOT be called with an invalid destination.

    FAILS: Current move route only checks self-destination, not descendants.
    """
    auth = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, auth["user"].id)

    # Pre-seed a folder item and a child subfolder
    parent_folder_id = "parent_folder_id"
    child_folder_id = "child_folder_id"
    from db.models import CloudItem
    parent_folder = CloudItem(
        id=_uuid.uuid4(),
        user_id=auth["user"].id,
        connection_id=conn.id,
        provider_item_id=parent_folder_id,
        name="Parent Folder",
        kind="folder",
        parent_ref=None,
    )
    child_folder = CloudItem(
        id=_uuid.uuid4(),
        user_id=auth["user"].id,
        connection_id=conn.id,
        provider_item_id=child_folder_id,
        name="Child Subfolder",
        kind="folder",
        parent_ref=parent_folder_id,
    )
    db_session.add(parent_folder)
    db_session.add(child_folder)
    await db_session.commit()

    # Try to move parent folder into its own child (descendant destination)
    payload = {
        "destination_parent_ref": child_folder_id,  # descendant of source
        "etag": "v1",
    }
    resp = await async_client.post(
        f"/api/cloud/connections/{conn.id}/items/{parent_folder_id}/move",
        headers=auth["headers"],
        json=payload,
    )
    assert resp.status_code in (400, 409, 422), (
        f"Expected rejection for descendant destination, got {resp.status_code}: {resp.text}"
    )
    body = resp.json()
    assert body.get("kind") == "invalid_destination", (
        f"Expected kind='invalid_destination', got {body.get('kind')!r}"
    )


async def test_move_success_upserts_cloud_item_before_returning(async_client, db_session):
    """POST move success upserts the moved item's new parent_ref in cloud_items before returning.

    Plan 09 Task 1 behavior 3: Successful move must update navigation metadata through
    centralized reconciliation (cloud_items.upsert_cloud_item) before the route returns
    success. The moved item's CloudItem row must reflect the new parent_ref.

    FAILS: Current move route does not call upsert_cloud_item on success.
    """
    from storage.cloud_base import MUT_KIND_UPDATED, MUT_REASON_MOVED
    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)

    move_item_id = "move_reconcile_item"
    original_parent_ref = "move_source_folder"
    dest_parent_ref = "move_dest_folder"

    # Pre-seed the item to be moved
    existing_item = CloudItem(
        id=_uuid.uuid4(),
        user_id=auth["user"].id,
        connection_id=conn.id,
        provider_item_id=move_item_id,
        name="Moving File.pdf",
        kind="file",
        parent_ref=original_parent_ref,
    )
    db_session.add(existing_item)
    await db_session.commit()

    mock_adapter = _make_mock_mutable_adapter(
        move_result={
            "kind": MUT_KIND_UPDATED,
            "reason": MUT_REASON_MOVED,
            "provider_item_id": move_item_id,
            "destination_parent_ref": dest_parent_ref,
        }
    )
    payload = {"destination_parent_ref": dest_parent_ref, "etag": "v1"}

    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}/items/{move_item_id}/move",
            headers=auth["headers"],
            json=payload,
        )

    assert resp.status_code == 200, f"Expected 200 for move, got {resp.status_code}: {resp.text}"
    body = resp.json()
    assert body.get("kind") == "moved"

    # After successful move, the CloudItem must have the new parent_ref
    result = await db_session.execute(
        select(CloudItem).where(
            CloudItem.connection_id == conn.id,
            CloudItem.provider_item_id == move_item_id,
        )
    )
    cloud_item = result.scalar_one_or_none()
    assert cloud_item is not None, "CloudItem must still exist after move"
    assert cloud_item.parent_ref == dest_parent_ref, (
        f"Moved CloudItem must have new parent_ref={dest_parent_ref!r}, "
        f"got {cloud_item.parent_ref!r} (Plan 09 Task 1 behavior 3)"
    )


async def test_move_success_marks_source_and_dest_folders_non_fresh(async_client, db_session):
    """POST move success marks both source and destination folders as non-fresh.

    Plan 09 Task 1 behavior 3: After a successful move, both the source folder
    (item removed) and the destination folder (item added) listings have changed.
    Both folder states must be invalidated so the next browse triggers a provider re-list.

    FAILS: Current move route does not update folder state on success.
    """
    from storage.cloud_base import MUT_KIND_UPDATED, MUT_REASON_MOVED
    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)

    move_item_id = "move_fresh_item"
    source_parent_ref = "move_fresh_source"
    dest_parent_ref = "move_fresh_dest"

    existing_item = CloudItem(
        id=_uuid.uuid4(),
        user_id=auth["user"].id,
        connection_id=conn.id,
        provider_item_id=move_item_id,
        name="File To Move.pdf",
        kind="file",
        parent_ref=source_parent_ref,
    )
    db_session.add(existing_item)
    await db_session.commit()

    mock_adapter = _make_mock_mutable_adapter(
        move_result={
            "kind": MUT_KIND_UPDATED,
            "reason": MUT_REASON_MOVED,
            "provider_item_id": move_item_id,
            "destination_parent_ref": dest_parent_ref,
        }
    )
    payload = {"destination_parent_ref": dest_parent_ref, "etag": "v1"}

    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}/items/{move_item_id}/move",
            headers=auth["headers"],
            json=payload,
        )

    assert resp.status_code == 200
    assert resp.json().get("kind") == "moved"

    # Source folder must be invalidated
    src_result = await db_session.execute(
        select(CloudFolderState).where(
            CloudFolderState.connection_id == conn.id,
            CloudFolderState.parent_ref == source_parent_ref,
        )
    )
    src_fs = src_result.scalar_one_or_none()
    assert src_fs is not None, (
        "Move success must invalidate the source folder CloudFolderState "
        "(Plan 09 Task 1 behavior 3)"
    )

    # Destination folder must also be invalidated
    dst_result = await db_session.execute(
        select(CloudFolderState).where(
            CloudFolderState.connection_id == conn.id,
            CloudFolderState.parent_ref == dest_parent_ref,
        )
    )
    dst_fs = dst_result.scalar_one_or_none()
    assert dst_fs is not None, (
        "Move success must also invalidate the destination folder CloudFolderState "
        "(Plan 09 Task 1 behavior 3)"
    )


async def test_move_success_writes_metadata_only_audit_row(async_client, db_session):
    """POST move success writes audit row 'cloud.item_moved' with metadata only.

    Plan 09 Task 1 behavior 3: Successful move must be audited with metadata only —
    no bytes, no provider URL, no tokens. Metadata must include connection_id and
    destination_parent_ref. The audit row is written before success returns.

    FAILS: Current move route does not write audit rows.
    """
    from storage.cloud_base import MUT_KIND_UPDATED, MUT_REASON_MOVED
    from db.models import AuditLog
    from sqlalchemy import select as sa_select, desc

    auth = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, auth["user"].id)

    mock_adapter = _make_mock_mutable_adapter(
        move_result={
            "kind": MUT_KIND_UPDATED,
            "reason": MUT_REASON_MOVED,
            "provider_item_id": "audit_move_item",
            "destination_parent_ref": "audit_dest_folder",
        }
    )
    payload = {"destination_parent_ref": "audit_dest_folder_unique", "etag": "v1"}

    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}/items/audit_move_item/move",
            headers=auth["headers"],
            json=payload,
        )

    assert resp.status_code == 200, f"Expected 200 for move, got {resp.status_code}"
    assert resp.json().get("kind") == "moved"

    # Audit row must be written before success returns
    rows_result = await db_session.execute(
        sa_select(AuditLog).where(
            AuditLog.user_id == auth["user"].id,
            AuditLog.event_type == "cloud.item_moved",
        ).order_by(desc(AuditLog.id)).limit(5)
    )
    rows = rows_result.scalars().all()
    assert len(rows) >= 1, (
        "Successful move must write audit row 'cloud.item_moved' before returning "
        "(Plan 09 Task 1 behavior 3)"
    )
    row = rows[0]
    meta = row.metadata_ or {}
    # No credentials allowed in audit metadata
    for forbidden in ("access_token", "refresh_token", "credentials_enc", "client_secret"):
        assert forbidden not in str(meta), (
            f"Move audit metadata must not contain '{forbidden}' (T-13-02)"
        )
    # Must have basic metadata
    assert "connection_id" in meta or "provider_item_id" in meta, (
        "Move audit metadata must include connection_id or provider_item_id"
    )


# ── Phase 13 Plan 09 Task 2 (RED): Delete — folder disclosure, metadata-only audit ─


async def test_delete_file_marks_parent_folder_non_fresh(async_client, db_session):
    """DELETE success marks the parent folder as non-fresh (reconcile-before-return).

    Plan 09 Task 2 behavior 1: After a successful delete, the parent folder listing
    has changed. The folder state must be invalidated so the next browse re-lists.

    FAILS: Current delete route does not update folder state on success.
    """
    from storage.cloud_base import MUT_KIND_DELETED, MUT_REASON_TRASHED
    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)

    delete_item_id = "delete_fresh_item"
    parent_ref = "delete_fresh_parent"

    existing_item = CloudItem(
        id=_uuid.uuid4(),
        user_id=auth["user"].id,
        connection_id=conn.id,
        provider_item_id=delete_item_id,
        name="To Be Deleted.pdf",
        kind="file",
        parent_ref=parent_ref,
    )
    db_session.add(existing_item)
    await db_session.commit()

    mock_adapter = _make_mock_mutable_adapter(
        delete_result={
            "kind": MUT_KIND_DELETED,
            "reason": MUT_REASON_TRASHED,
            "provider_item_id": delete_item_id,
        }
    )

    with patch("storage.cloud_backend_factory.build_mutable_cloud_adapter", return_value=mock_adapter):
        resp = await async_client.delete(
            f"/api/cloud/connections/{conn.id}/items/{delete_item_id}",
            headers=auth["headers"],
        )

    assert resp.status_code in (200, 204), (
        f"Expected 200/204 for delete, got {resp.status_code}: {resp.text}"
    )
    if resp.status_code == 200:
        assert resp.json().get("kind") == "deleted"

    # Parent folder must be invalidated after delete
    result = await db_session.execute(
        select(CloudFolderState).where(
            CloudFolderState.connection_id == conn.id,
            CloudFolderState.parent_ref == parent_ref,
        )
    )
    fs = result.scalar_one_or_none()
    assert fs is not None, (
        "Delete success must create/update a CloudFolderState row for the parent folder "
        "(Plan 09 Task 2 behavior 1)"
    )


async def test_delete_folder_returns_stronger_disclosure_than_file(async_client, db_session):
    """DELETE folder returns stronger disclosure message than file delete (D-10).

    Plan 09 Task 2 behavior 1: Folder delete must warn that nested contents are included.
    The response body must carry folder_delete=True or a stronger reason code to distinguish
    folder vs file deletion disclosure.

    FAILS: Current delete route returns identical body regardless of item kind.
    """
    from storage.cloud_base import MUT_KIND_DELETED, MUT_REASON_TRASHED
    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)

    folder_item_id = "delete_folder_disclosure"
    folder_item = CloudItem(
        id=_uuid.uuid4(),
        user_id=auth["user"].id,
        connection_id=conn.id,
        provider_item_id=folder_item_id,
        name="Folder To Delete",
        kind="folder",
        parent_ref=None,
    )
    db_session.add(folder_item)
    await db_session.commit()

    mock_adapter = _make_mock_mutable_adapter(
        delete_result={
            "kind": MUT_KIND_DELETED,
            "reason": MUT_REASON_TRASHED,
            "provider_item_id": folder_item_id,
        }
    )

    with patch("storage.cloud_backend_factory.build_mutable_cloud_adapter", return_value=mock_adapter):
        resp = await async_client.delete(
            f"/api/cloud/connections/{conn.id}/items/{folder_item_id}",
            headers=auth["headers"],
        )

    assert resp.status_code in (200, 204), (
        f"Expected 200/204 for folder delete, got {resp.status_code}: {resp.text}"
    )
    if resp.status_code == 200:
        body = resp.json()
        assert body.get("kind") == "deleted", f"Expected kind='deleted', got {body.get('kind')!r}"
        # Folder delete must have stronger disclosure — either is_folder=True or
        # a specific reason that indicates folder semantics
        assert body.get("is_folder") is True or body.get("item_kind") == "folder", (
            "Folder delete response must disclose that it is a folder (D-10): "
            f"body={body!r}"
        )


async def test_delete_success_writes_metadata_only_audit_row(async_client, db_session):
    """DELETE success writes audit row 'cloud.item_deleted' with metadata only.

    Plan 09 Task 2 behavior 3: Successful delete must be audited before success returns.
    Metadata must include connection_id, provider_item_id, and delete kind (trashed/permanent).
    No provider URL, token, or bytes may appear in the audit row.

    FAILS: Current delete route does not write audit rows.
    """
    from storage.cloud_base import MUT_KIND_DELETED, MUT_REASON_TRASHED
    from db.models import AuditLog
    from sqlalchemy import select as sa_select, desc

    auth = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, auth["user"].id)

    mock_adapter = _make_mock_mutable_adapter(
        delete_result={
            "kind": MUT_KIND_DELETED,
            "reason": MUT_REASON_TRASHED,
            "provider_item_id": "audit_delete_item",
        }
    )

    with patch("storage.cloud_backend_factory.build_mutable_cloud_adapter", return_value=mock_adapter):
        resp = await async_client.delete(
            f"/api/cloud/connections/{conn.id}/items/audit_delete_item_unique",
            headers=auth["headers"],
        )

    assert resp.status_code in (200, 204), (
        f"Expected 200/204 for delete, got {resp.status_code}: {resp.text}"
    )

    # Audit row must be written before success returns
    result = await db_session.execute(
        sa_select(AuditLog).where(
            AuditLog.user_id == auth["user"].id,
            AuditLog.event_type == "cloud.item_deleted",
        ).order_by(desc(AuditLog.id)).limit(5)
    )
    rows = result.scalars().all()
    assert len(rows) >= 1, (
        "Successful delete must write audit row 'cloud.item_deleted' before returning "
        "(Plan 09 Task 2 behavior 3)"
    )
    row = rows[0]
    meta = row.metadata_ or {}
    # No credentials in audit metadata
    for forbidden in ("access_token", "refresh_token", "credentials_enc", "client_secret"):
        assert forbidden not in str(meta), (
            f"Delete audit metadata must not contain '{forbidden}' (T-13-02)"
        )
    # Must indicate trash vs permanent
    assert "delete_kind" in meta or "reason" in meta, (
        "Delete audit metadata must indicate 'trashed' vs 'permanent' (D-11)"
    )


async def test_delete_failed_does_not_write_false_audit_event(async_client, db_session):
    """Failed delete (offline) must NOT write a false 'cloud.item_deleted' audit event.

    Plan 09 Task 2 behavior 3: Only authoritative delete success may emit audit rows.
    An offline or reauth_required result must not log a phantom delete event.

    FAILS: Establishes invariant — current delete does not write any audit rows yet.
    """
    from storage.cloud_base import MUT_KIND_OFFLINE, MUT_REASON_PROVIDER_OFFLINE
    from db.models import AuditLog
    from sqlalchemy import select as sa_select, desc

    auth = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, auth["user"].id)

    # Count audit rows before the failed delete
    before_result = await db_session.execute(
        sa_select(AuditLog).where(
            AuditLog.user_id == auth["user"].id,
            AuditLog.event_type == "cloud.item_deleted",
        )
    )
    before_count = len(before_result.scalars().all())

    mock_adapter = _make_mock_mutable_adapter(
        delete_result={"kind": MUT_KIND_OFFLINE, "reason": MUT_REASON_PROVIDER_OFFLINE}
    )

    with patch("storage.cloud_backend_factory.build_mutable_cloud_adapter", return_value=mock_adapter):
        resp = await async_client.delete(
            f"/api/cloud/connections/{conn.id}/items/offline_delete_item",
            headers=auth["headers"],
        )

    assert resp.status_code != 500, (
        "Offline delete must return typed body, not 500"
    )

    # No phantom audit row must be created
    after_result = await db_session.execute(
        sa_select(AuditLog).where(
            AuditLog.user_id == auth["user"].id,
            AuditLog.event_type == "cloud.item_deleted",
        )
    )
    after_count = len(after_result.scalars().all())
    assert after_count == before_count, (
        f"Failed (offline) delete must not write a false 'cloud.item_deleted' audit row "
        f"(Plan 09 Task 2 behavior 3): count changed {before_count} → {after_count}"
    )