test(13-09): add failing RED tests for move stale-guard, reconciliation, and delete disclosure/audit

- Move: stale etag returns 409 + refreshes source folder state
- Move: descendant destination rejected (D-09)
- Move: success upserts cloud_item with new parent_ref (reconcile-before-return)
- Move: success marks source + destination folders non-fresh
- Move: success writes metadata-only 'cloud.item_moved' audit row
- Delete: success marks parent folder non-fresh
- Delete: folder disclosure stronger than file (is_folder=True, D-10)
- Delete: success writes metadata-only 'cloud.item_deleted' audit row
- Delete: failed delete must not write false audit event

backend/tests/test_cloud_mutations.py

@@ -1801,3 +1801,576 @@ async def test_rename_failed_does_not_mutate_cloud_items(async_client, db_sessio
         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}"
+    )