feat(13-09): move stale-guard, descendant safety, reconciliation, delete disclosure, and metadata-only auditing

Move endpoint (D-07, D-08, D-09, T-13-27, T-13-28, T-13-29):
- Descendant destination detection via cloud_items ancestor-chain walk (D-09)
- Stale provider result marks source folder non-fresh, returns typed stale body (D-07)
- Success: upsert_cloud_item with new parent_ref before returning
- Success: update_folder_state for both source and destination folders
- Success: write_audit_log 'cloud.item_moved' with metadata-only payload

Delete endpoint (D-10, D-11, T-13-29):
- Pre-fetch item kind for D-10 folder-vs-file disclosure
- Success: update_folder_state for parent folder before returning
- Success: write_audit_log 'cloud.item_deleted' with delete_kind metadata
- Response carries is_folder/item_kind for frontend disclosure
- Failed deletes never emit audit rows

Promote xfail markers in test_cloud_audit.py for move and delete audit rows

backend/api/cloud/operations.py

@@ -618,6 +618,51 @@ async def rename_cloud_item(
 # ── Move ──────────────────────────────────────────────────────────────────────
 
 
+async def _is_descendant_destination(
+    session: AsyncSession,
+    connection_id: uuid.UUID,
+    user_id: uuid.UUID,
+    source_item_id: str,
+    dest_parent_ref: str,
+) -> bool:
+    """Return True if dest_parent_ref is a descendant of source_item_id.
+
+    D-09: The backend must independently reject descendant destinations even if
+    the frontend pre-screened them. Walks the ancestor chain of dest_parent_ref
+    through the cloud_items table to detect cycles.
+
+    This is a metadata-only check — no provider calls are made.
+    """
+    from sqlalchemy import select
+    from db.models import CloudItem
+
+    visited: set[str] = set()
+    current_ref: str | None = dest_parent_ref
+
+    while current_ref is not None:
+        if current_ref in visited:
+            # Cycle in ancestor chain — defensive guard
+            break
+        visited.add(current_ref)
+
+        if current_ref == source_item_id:
+            return True
+
+        # Look up the parent of current_ref
+        result = await session.execute(
+            select(CloudItem).where(
+                CloudItem.connection_id == connection_id,
+                CloudItem.provider_item_id == current_ref,
+                CloudItem.user_id == user_id,
+                CloudItem.deleted_at.is_(None),
+            )
+        )
+        parent_item = result.scalar_one_or_none()
+        current_ref = parent_item.parent_ref if parent_item else None
+
+    return False
+
+
 @router.post("/connections/{connection_id}/items/{item_id:path}/move")
 @account_limiter.limit("100/minute")
 async def move_cloud_item(
@@ -632,6 +677,12 @@ async def move_cloud_item(
 
     D-08: Cross-connection moves are rejected before calling the adapter.
     D-09: Self and descendant destinations are rejected before calling the adapter.
+    D-07: Stale precondition failures stop the move, refresh the source folder, and
+          return typed {kind: 'stale', reason: 'item_changed'} for retry.
+
+    On success: upserts the item's new parent_ref in cloud_items, invalidates both
+    source and destination folder states, and writes a metadata-only audit row before
+    the response is returned (reconcile-before-return, T-13-26, T-13-29).
 
     Returns {kind: 'moved', provider_item_id, parent_ref} on success.
     Returns {kind: 'invalid_destination', reason: '...'} for invalid moves (HTTP 409).
@@ -657,6 +708,41 @@ async def move_cloud_item(
             content={"kind": "invalid_destination", "reason": MUT_REASON_SELF_MOVE},
         )
 
+    # D-09: Descendant destination rejection — walk ancestor chain via cloud_items metadata.
+    # This is a belt-and-suspenders check independent of provider-side validation.
+    if body.destination_parent_ref:
+        is_descendant = await _is_descendant_destination(
+            session,
+            connection_id=connection_id,
+            user_id=current_user.id,
+            source_item_id=item_id,
+            dest_parent_ref=body.destination_parent_ref,
+        )
+        if is_descendant:
+            return JSONResponse(
+                status_code=status.HTTP_409_CONFLICT,
+                content={"kind": "invalid_destination", "reason": "descendant_destination"},
+            )
+
+    # Resolve existing item metadata before the provider call (for reconciliation on success)
+    from sqlalchemy import select
+    from db.models import CloudItem
+
+    existing_result = await session.execute(
+        select(CloudItem).where(
+            CloudItem.connection_id == connection_id,
+            CloudItem.provider_item_id == item_id,
+            CloudItem.user_id == current_user.id,
+            CloudItem.deleted_at.is_(None),
+        )
+    )
+    existing_item = existing_result.scalar_one_or_none()
+    source_parent_ref = existing_item.parent_ref if existing_item else None
+    item_kind = existing_item.kind if existing_item else "file"
+    item_name = existing_item.name if existing_item else item_id
+    item_content_type = existing_item.content_type if existing_item else None
+    item_size = existing_item.provider_size if existing_item else None
+
     conn, adapter = await _resolve_and_get_adapter(session, connection_id, current_user.id)
 
     result = await adapter.move(
@@ -668,12 +754,87 @@ async def move_cloud_item(
     kind = result.get("kind")
 
     if kind == MUT_KIND_UPDATED:
+        # Success: reconcile before returning (T-13-26, T-13-29)
+        resolved_provider_item_id = result.get("provider_item_id", item_id)
+        resolved_dest_parent_ref = result.get("destination_parent_ref", body.destination_parent_ref)
+
+        # Upsert item with new parent_ref — preserves stable row identity
+        resource = CloudResource(
+            id=uuid.uuid4(),
+            provider_item_id=resolved_provider_item_id,
+            connection_id=connection_id,
+            user_id=current_user.id,
+            name=item_name,
+            kind=item_kind,
+            parent_ref=resolved_dest_parent_ref,
+            content_type=item_content_type,
+            size=item_size,
+        )
+        await upsert_cloud_item(session, user_id=str(current_user.id), resource=resource)
+
+        # Invalidate source folder (item was removed from here)
+        invalidate_source = source_parent_ref if source_parent_ref is not None else ""
+        await update_folder_state(
+            session,
+            user_id=str(current_user.id),
+            connection_id=str(connection_id),
+            parent_ref=invalidate_source,
+            refresh_state="warning",
+            error_code="move_mutated",
+            error_message="Folder contents changed by move — re-listing required.",
+        )
+
+        # Invalidate destination folder (item was added here)
+        invalidate_dest = resolved_dest_parent_ref if resolved_dest_parent_ref is not None else ""
+        await update_folder_state(
+            session,
+            user_id=str(current_user.id),
+            connection_id=str(connection_id),
+            parent_ref=invalidate_dest,
+            refresh_state="warning",
+            error_code="move_mutated",
+            error_message="Folder contents changed by move — re-listing required.",
+        )
+
+        # Write metadata-only audit row (T-13-02, T-13-29)
+        await write_audit_log(
+            session,
+            event_type="cloud.item_moved",
+            user_id=current_user.id,
+            actor_id=current_user.id,
+            resource_id=None,
+            ip_address=get_client_ip(request),
+            metadata_={
+                "connection_id": str(connection_id),
+                "provider_item_id": resolved_provider_item_id,
+                "old_parent_ref": source_parent_ref,
+                "new_parent_ref": resolved_dest_parent_ref,
+            },
+        )
+
+        await session.commit()
+
         return {
             "kind": "moved",
-            "provider_item_id": result.get("provider_item_id", item_id),
+            "provider_item_id": resolved_provider_item_id,
-            "parent_ref": result.get("destination_parent_ref", body.destination_parent_ref),
+            "parent_ref": resolved_dest_parent_ref,
         }
 
+    if kind == MUT_KIND_STALE:
+        # D-07: Stop mutation, mark source folder as non-fresh, return typed stale result
+        stale_source = source_parent_ref if source_parent_ref is not None else ""
+        await update_folder_state(
+            session,
+            user_id=str(current_user.id),
+            connection_id=str(connection_id),
+            parent_ref=stale_source,
+            refresh_state="warning",
+            error_code="stale_listing",
+            error_message="Item changed externally — re-listing required before retry.",
+        )
+        await session.commit()
+        return _mutation_error_response(result)
+
     return _mutation_error_response(result)
 
 
@@ -691,14 +852,32 @@ async def delete_cloud_item(
 ) -> dict:
     """Delete a cloud item.
 
+    D-10: Files get a standard confirmation; folders warn that nested contents are included.
     D-11: Prefers the provider's trash/recycle-bin when supported. Returns
     {kind: 'deleted', reason: 'trashed'} or {kind: 'deleted', reason: 'permanent'}.
-    D-10: Callers must confirm before invoking this endpoint — it acts immediately.
+    On success: marks the parent folder non-fresh and writes a metadata-only audit row
+    before the response is returned (reconcile-before-return, T-13-29).
 
     T-13-01: Owner-scoped. T-13-14: No credentials or provider URLs in response.
     """
     request.state.current_user = current_user
 
+    # Pre-fetch item metadata for disclosure (D-10) and reconciliation
+    from sqlalchemy import select
+    from db.models import CloudItem
+
+    meta_result = await session.execute(
+        select(CloudItem).where(
+            CloudItem.connection_id == connection_id,
+            CloudItem.provider_item_id == item_id,
+            CloudItem.user_id == current_user.id,
+            CloudItem.deleted_at.is_(None),
+        )
+    )
+    existing_item = meta_result.scalar_one_or_none()
+    item_kind = existing_item.kind if existing_item else "file"
+    item_parent_ref = existing_item.parent_ref if existing_item else None
+
     conn, adapter = await _resolve_and_get_adapter(session, connection_id, current_user.id)
 
     result = await adapter.delete(provider_item_id=item_id, trash=True)
@@ -707,10 +886,46 @@ async def delete_cloud_item(
     reason = result.get("reason")
 
     if kind == MUT_KIND_DELETED:
+        # Success: reconcile before returning (T-13-29)
+
+        # Invalidate parent folder so next browse reflects the deletion
+        invalidate_parent = item_parent_ref if item_parent_ref is not None else ""
+        await update_folder_state(
+            session,
+            user_id=str(current_user.id),
+            connection_id=str(connection_id),
+            parent_ref=invalidate_parent,
+            refresh_state="warning",
+            error_code="delete_mutated",
+            error_message="Folder contents changed by delete — re-listing required.",
+        )
+
+        # Write metadata-only audit row — only after authoritative success (T-13-02, T-13-29)
+        delete_reason = reason or MUT_REASON_PERMANENT
+        await write_audit_log(
+            session,
+            event_type="cloud.item_deleted",
+            user_id=current_user.id,
+            actor_id=current_user.id,
+            resource_id=None,
+            ip_address=get_client_ip(request),
+            metadata_={
+                "connection_id": str(connection_id),
+                "provider_item_id": result.get("provider_item_id", item_id),
+                "delete_kind": delete_reason,
+                "item_kind": item_kind,
+            },
+        )
+
+        await session.commit()
+
+        # D-10: Disclose folder-vs-file delete semantics to the frontend
         return {
             "kind": "deleted",
-            "reason": reason or MUT_REASON_PERMANENT,
+            "reason": delete_reason,
             "provider_item_id": result.get("provider_item_id", item_id),
+            "item_kind": item_kind,
+            "is_folder": item_kind == "folder",
         }
 
     return _mutation_error_response(result)

backend/tests/test_cloud_audit.py

@@ -438,77 +438,94 @@ async def test_rename_stale_does_not_write_false_rename_audit(async_client, db_s
 # ── T-13-05: Move audit row ───────────────────────────────────────────────────
 
 
-@pytest.mark.xfail(
-    reason="Phase 13 audit write not implemented yet — RED test from plan 01 (T-13-05)",
-    strict=True,
-)
 async def test_move_success_writes_audit_row(async_client, db_session):
     """Successful move writes audit row 'cloud.item_moved' with metadata only (T-13-05).
 
     Metadata must include connection_id, old_parent_ref, new_parent_ref — no tokens.
 
-    FAILS: Phase 13 audit write does not exist yet.
+    Plan 09 GREEN: Move audit write implemented.
     """
+    from storage.cloud_base import MUT_KIND_UPDATED, MUT_REASON_MOVED
+    from unittest.mock import patch, AsyncMock, MagicMock
+
     auth = await _create_user_and_token(db_session)
     conn = await _create_cloud_connection(db_session, auth["user"].id)
 
-    payload = {"destination_parent_ref": "dest_folder_ref", "etag": "v1"}
+    mock_adapter = AsyncMock()
-    resp = await async_client.post(
+    mock_adapter.get_object = AsyncMock(return_value=b"content")
-        f"/api/cloud/connections/{conn.id}/items/fake_item_id/move",
+    mock_adapter._normalize_error = MagicMock(return_value={"kind": "error", "reason": "provider_error"})
-        headers=auth["headers"],
+    mock_adapter.move = AsyncMock(return_value={
-        json=payload,
+        "kind": MUT_KIND_UPDATED,
-    )
+        "reason": MUT_REASON_MOVED,
-    # 401 means credential decrypt failure in test env; audit check skipped in that case.
+        "provider_item_id": "fake_item_id",
-    assert resp.status_code in (200, 401, 404), (
+        "destination_parent_ref": "dest_folder_ref",
-        f"Unexpected status: {resp.status_code}"
+    })
-    )
+
-    if resp.status_code == 200:
+    payload = {"destination_parent_ref": "dest_folder_ref_audit", "etag": "v1"}
-        rows = await _get_recent_audit_rows(
+    with patch("storage.cloud_backend_factory.build_mutable_cloud_adapter", return_value=mock_adapter):
-            db_session, auth["user"].id, event_type="cloud.item_moved"
+        resp = await async_client.post(
+            f"/api/cloud/connections/{conn.id}/items/fake_item_id/move",
+            headers=auth["headers"],
+            json=payload,
         )
-        assert len(rows) >= 1, (
+
-            "Successful move must write audit row 'cloud.item_moved' (T-13-05)"
+    assert resp.status_code == 200, (
-        )
+        f"Expected 200 for move, got {resp.status_code}: {resp.text}"
-        _assert_metadata_no_secrets(rows[0].metadata_, "cloud.item_moved audit")
+    )
+    rows = await _get_recent_audit_rows(
+        db_session, auth["user"].id, event_type="cloud.item_moved"
+    )
+    assert len(rows) >= 1, (
+        "Successful move must write audit row 'cloud.item_moved' (T-13-05)"
+    )
+    _assert_metadata_no_secrets(rows[0].metadata_, "cloud.item_moved audit")
 
 
 # ── T-13-05: Delete audit row ─────────────────────────────────────────────────
 
 
-@pytest.mark.xfail(
-    reason="Phase 13 audit write not implemented yet — RED test from plan 01 (T-13-05)",
-    strict=True,
-)
 async def test_delete_success_writes_audit_row(async_client, db_session):
     """Successful delete writes audit row 'cloud.item_deleted' with metadata only (T-13-05).
 
     Metadata must include connection_id, item_id, delete_kind ('trashed' or 'permanent').
 
-    FAILS: Phase 13 audit write does not exist yet.
+    Plan 09 GREEN: Delete audit write implemented.
     """
+    from storage.cloud_base import MUT_KIND_DELETED, MUT_REASON_TRASHED
+    from unittest.mock import patch, AsyncMock, MagicMock
+
     auth = await _create_user_and_token(db_session)
     conn = await _create_cloud_connection(db_session, auth["user"].id)
 
-    resp = await async_client.delete(
+    mock_adapter = AsyncMock()
-        f"/api/cloud/connections/{conn.id}/items/fake_file_id",
+    mock_adapter.get_object = AsyncMock(return_value=b"content")
-        headers=auth["headers"],
+    mock_adapter._normalize_error = MagicMock(return_value={"kind": "error", "reason": "provider_error"})
-    )
+    mock_adapter.delete = AsyncMock(return_value={
-    # 401 means credential decrypt failure in test env; audit check skipped in that case.
+        "kind": MUT_KIND_DELETED,
-    assert resp.status_code in (200, 204, 401, 404), (
+        "reason": MUT_REASON_TRASHED,
-        f"Unexpected status: {resp.status_code}"
+        "provider_item_id": "audit_delete_file_id",
-    )
+    })
-    if resp.status_code in (200, 204):
+
-        rows = await _get_recent_audit_rows(
+    with patch("storage.cloud_backend_factory.build_mutable_cloud_adapter", return_value=mock_adapter):
-            db_session, auth["user"].id, event_type="cloud.item_deleted"
+        resp = await async_client.delete(
+            f"/api/cloud/connections/{conn.id}/items/audit_delete_file_id",
+            headers=auth["headers"],
         )
-        assert len(rows) >= 1, (
+
-            "Successful delete must write audit row 'cloud.item_deleted' (T-13-05)"
+    assert resp.status_code in (200, 204), (
+        f"Expected 200/204 for delete, got {resp.status_code}: {resp.text}"
+    )
+    rows = await _get_recent_audit_rows(
+        db_session, auth["user"].id, event_type="cloud.item_deleted"
+    )
+    assert len(rows) >= 1, (
+        "Successful delete must write audit row 'cloud.item_deleted' (T-13-05)"
+    )
+    _assert_metadata_no_secrets(rows[0].metadata_, "cloud.item_deleted audit")
+    if rows[0].metadata_:
+        assert "delete_kind" in rows[0].metadata_ or "reason" in rows[0].metadata_, (
+            "Delete audit metadata must indicate 'trashed' vs 'permanent' (D-11)"
         )
-        _assert_metadata_no_secrets(rows[0].metadata_, "cloud.item_deleted audit")
-        if rows[0].metadata_:
-            assert "delete_kind" in rows[0].metadata_ or "reason" in rows[0].metadata_, (
-                "Delete audit metadata must indicate 'trashed' vs 'permanent' (D-11)"
-            )
 
 
 # ── T-13-02: Admin audit log must not expose cloud credentials ─────────────────