feat(06.2-03): backend — cloud-aware delete routing + skip_quota + remove_only param

- storage.delete_document gains skip_quota=False param; quota decrement gated on it
- DELETE /api/documents/{id} gains remove_only=bool query param
- Cloud docs (storage_backend != minio): attempt cloud backend delete_object first
  - On failure: return HTTP 200 {success: false, cloud_delete_failed: true} (not 4xx)
  - On success or remove_only: delete DB row with skip_quota=True
- Cloud creds/exception message never included in response body (T-06.2-03-02)
- Promote 3 xfail stubs to real tests (propagates, failure, remove_only)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

backend/api/documents.py

@@ -26,7 +26,7 @@ from pathlib import Path
 from typing import Optional
 
 from fastapi import APIRouter, Depends, Form, HTTPException, Query, Request, UploadFile, File, status
-from fastapi.responses import StreamingResponse
+from fastapi.responses import JSONResponse, StreamingResponse
 from pydantic import BaseModel, Field, field_validator
 from sqlalchemy import select, text, func
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -605,13 +605,18 @@ async def patch_document(
 async def delete_document(
     doc_id: str,
     request: Request,
+    remove_only: bool = Query(default=False),
     session: AsyncSession = Depends(get_db),
     current_user: User = Depends(get_regular_user),
 ):
     """Delete a document and decrement quota atomically.
 
-    services.storage.delete_document handles the atomic quota decrement
+    For cloud-stored documents:
-    (STORE-06, D-07) via GREATEST(0, used_bytes - delta) SQL.
+    - Default path: attempt cloud provider delete first; on failure return
+      {success: false, cloud_delete_failed: true} (HTTP 200) so the frontend
+      can offer a "Remove from app" fallback (T-06.2-03-02).
+    - remove_only=true: skip cloud delete, remove DB row only, skip quota decrement.
+    - Cloud docs always use skip_quota=True (never charged MinIO quota, T-06.2-03-01).
 
     D-16: requires authenticated regular user. Asserts ownership — cross-user
     delete returns 404 (not 403) to avoid information leakage (T-03-11).
@@ -625,12 +630,29 @@ async def delete_document(
     if doc is None or doc.user_id != current_user.id:
         raise HTTPException(404, "Document not found")
 
-    # Capture audit metadata before delete removes the row
+    is_cloud = doc.storage_backend != "minio"
     _doc_size = doc.size_bytes
     _doc_id = doc.id
     _ip = request.headers.get("X-Forwarded-For") or (request.client.host if request.client else None)
 
-    ok = await storage.delete_document(session, doc_id)
+    # Cloud routing: attempt provider delete unless remove_only is set
+    if is_cloud and not remove_only:
+        try:
+            cloud_backend = await get_storage_backend_for_document(doc, current_user, session)
+            await cloud_backend.delete_object(doc.object_key)
+        except Exception as exc:
+            import sys
+            print(f"[cloud-delete] provider error: {exc}", file=sys.stderr)
+            return JSONResponse(
+                status_code=200,
+                content={
+                    "success": False,
+                    "cloud_delete_failed": True,
+                    "detail": "Cloud provider delete failed. You can remove from app only.",
+                },
+            )
+
+    ok = await storage.delete_document(session, doc_id, skip_quota=is_cloud)
     if not ok:
         raise HTTPException(404, "Document not found")
 

backend/services/storage.py

@@ -140,12 +140,15 @@ async def list_metadata(
     return rows
 
 
-async def delete_document(session: AsyncSession, doc_id: str) -> bool:
+async def delete_document(session: AsyncSession, doc_id: str, skip_quota: bool = False) -> bool:
     """Delete a document's MinIO object and its PostgreSQL row.
 
     Returns False if the document is not found; True on success.
     MinIO deletion failures are logged to stderr but do not prevent the DB row
     deletion (the bytes may already be gone).
+
+    skip_quota=True skips the quota decrement — used for cloud-stored documents
+    that were never charged against the user's MinIO quota (T-06.2-03-01).
     """
     try:
         uid = uuid.UUID(doc_id)
@@ -161,18 +164,19 @@ async def delete_document(session: AsyncSession, doc_id: str) -> bool:
     except Exception as exc:
         print(f"[storage] WARNING: MinIO delete_object failed for {doc.object_key!r}: {exc}", file=sys.stderr)
 
-    # Atomic quota decrement (STORE-06, D-07).
+    if not skip_quota:
-    # user_id is always set post-migration (Plan 03-03+) — guard removed.
+        # Atomic quota decrement (STORE-06, D-07).
-    # Use CASE WHEN instead of GREATEST() for SQLite compatibility
+        # user_id is always set post-migration (Plan 03-03+) — guard removed.
-    # (PostgreSQL supports both; SQLite lacks the GREATEST scalar function).
+        # Use CASE WHEN instead of GREATEST() for SQLite compatibility
-    await session.execute(
+        # (PostgreSQL supports both; SQLite lacks the GREATEST scalar function).
-        text(
+        await session.execute(
-            "UPDATE quotas "
+            text(
-            "SET used_bytes = CASE WHEN used_bytes > :delta THEN used_bytes - :delta ELSE 0 END "
+                "UPDATE quotas "
-            "WHERE user_id = :uid"
+                "SET used_bytes = CASE WHEN used_bytes > :delta THEN used_bytes - :delta ELSE 0 END "
-        ),
+                "WHERE user_id = :uid"
-        {"delta": doc.size_bytes, "uid": str(doc.user_id)},
+            ),
-    )
+            {"delta": doc.size_bytes, "uid": str(doc.user_id)},
+        )
 
     await session.delete(doc)
     await session.commit()

backend/tests/test_documents.py

@@ -636,16 +636,106 @@ async def test_stream_document_content_cloud_backend_error(async_client, auth_us
 # ---------------------------------------------------------------------------
 
 
-async def test_delete_cloud_document_propagates(async_client, auth_user, db_session):
+async def test_delete_cloud_document_propagates(async_client, auth_user, db_session, monkeypatch):
     """DELETE /api/documents/{id} for a cloud doc calls cloud backend delete_object (D-01)"""
-    pytest.xfail("Phase 6.2 — not implemented yet")
+    import uuid as _uuid
+    from unittest.mock import AsyncMock
+    from db.models import Document
+
+    doc_id = _uuid.uuid4()
+    doc = Document(
+        id=doc_id,
+        user_id=auth_user["user"].id,
+        filename="gdrive_doc.pdf",
+        content_type="application/pdf",
+        size_bytes=512,
+        storage_backend="google_drive",
+        status="uploaded",
+        object_key=f"{auth_user['user'].id}/{doc_id}/{_uuid.uuid4()}.pdf",
+    )
+    db_session.add(doc)
+    await db_session.commit()
+
+    mock_backend = AsyncMock()
+    mock_backend.delete_object = AsyncMock(return_value=None)
+
+    async def fake_get_backend(*args, **kwargs):
+        return mock_backend
+
+    monkeypatch.setattr("api.documents.get_storage_backend_for_document", fake_get_backend)
+
+    resp = await async_client.delete(f"/api/documents/{doc_id}", headers=auth_user["headers"])
+    assert resp.status_code == 200, resp.text
+    assert resp.json()["success"] is True
+    mock_backend.delete_object.assert_called_once()
+
+    # DB row removed
+    deleted = await db_session.get(Document, doc_id)
+    assert deleted is None
 
 
-async def test_delete_cloud_document_failure(async_client, auth_user, db_session):
+async def test_delete_cloud_document_failure(async_client, auth_user, db_session, monkeypatch):
     """DELETE /api/documents/{id} returns cloud_delete_failed=True when provider raises (D-03)"""
-    pytest.xfail("Phase 6.2 — not implemented yet")
+    import uuid as _uuid
+    from unittest.mock import AsyncMock
+    from db.models import Document
+
+    doc_id = _uuid.uuid4()
+    doc = Document(
+        id=doc_id,
+        user_id=auth_user["user"].id,
+        filename="gdrive_fail.pdf",
+        content_type="application/pdf",
+        size_bytes=512,
+        storage_backend="google_drive",
+        status="uploaded",
+        object_key=f"{auth_user['user'].id}/{doc_id}/{_uuid.uuid4()}.pdf",
+    )
+    db_session.add(doc)
+    await db_session.commit()
+
+    async def raise_provider_error(*args, **kwargs):
+        raise RuntimeError("provider error")
+
+    monkeypatch.setattr("api.documents.get_storage_backend_for_document", raise_provider_error)
+
+    resp = await async_client.delete(f"/api/documents/{doc_id}", headers=auth_user["headers"])
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    assert body.get("cloud_delete_failed") is True, f"Expected cloud_delete_failed=True, got {body}"
+    assert body.get("success") is False
+
+    # DB row must NOT be deleted (soft-failure path)
+    still_there = await db_session.get(Document, doc_id)
+    assert still_there is not None, "DB row should not be deleted when cloud delete fails"
 
 
 async def test_delete_cloud_remove_only(async_client, auth_user, db_session):
     """DELETE /api/documents/{id}?remove_only=true skips cloud delete, removes DB row only (D-02)"""
-    pytest.xfail("Phase 6.2 — not implemented yet")
+    import uuid as _uuid
+    from db.models import Document, Quota
+
+    doc_id = _uuid.uuid4()
+    doc = Document(
+        id=doc_id,
+        user_id=auth_user["user"].id,
+        filename="gdrive_remove_only.pdf",
+        content_type="application/pdf",
+        size_bytes=1024,
+        storage_backend="google_drive",
+        status="uploaded",
+        object_key=f"{auth_user['user'].id}/{doc_id}/{_uuid.uuid4()}.pdf",
+    )
+    db_session.add(doc)
+    await db_session.commit()
+
+    resp = await async_client.delete(
+        f"/api/documents/{doc_id}?remove_only=true",
+        headers=auth_user["headers"],
+    )
+    assert resp.status_code == 200, resp.text
+    assert resp.json()["success"] is True
+
+    # DB row removed
+    deleted = await db_session.get(Document, doc_id)
+    assert deleted is None