feat(12.1-02): centralize listing freshness gate — apply_listing_and_finalize

- Add ListingResult dataclass to cloud_items.py
- Add apply_listing_and_finalize() — single shared gate for complete/incomplete semantics
- complete=True: upserts, soft-deletes unseen children, marks fresh, advances last_refreshed_at
- complete=False: upserts seen items only, never deletes unseen, preserves prior timestamp, warns
- browse.py: replace unconditional update_folder_state('fresh') with apply_listing_and_finalize
- cloud_tasks.py: replace unconditional fresh transition with apply_listing_and_finalize
- Celery task returns structured warning status on incomplete (no credentials/provider item names)
- Fix timezone naive/aware comparison in test (SQLite strips tzinfo)

backend/services/cloud_items.py

@@ -12,6 +12,7 @@ Domain exceptions:
 from __future__ import annotations
 
 import uuid
+from dataclasses import dataclass, field
 from datetime import datetime, timezone
 from typing import Optional, Sequence
 
@@ -250,6 +251,97 @@ async def get_or_create_folder_state(
     return fs
 
 
+# ── Listing application result ────────────────────────────────────────────────
+
+@dataclass
+class ListingResult:
+    """Structured result from apply_listing_and_finalize.
+
+    is_fresh: True only when the provider supplied a complete authoritative listing.
+    warning_code: stable controlled code when is_fresh is False (never raw exception text).
+    warning_message: stable controlled message for API consumers (no credentials/URLs).
+    """
+    is_fresh: bool
+    warning_code: Optional[str] = None
+    warning_message: Optional[str] = None
+
+
+# ── Single listing application gate ──────────────────────────────────────────
+
+async def apply_listing_and_finalize(
+    session: AsyncSession,
+    *,
+    user_id: str,
+    connection_id: str,
+    parent_ref: Optional[str],
+    listing: CloudListing,
+) -> ListingResult:
+    """Apply a provider listing and finalize CloudFolderState atomically.
+
+    This is the SINGLE shared gate used by both the synchronous browse endpoint
+    and the Celery background worker.  Neither caller may independently decide
+    freshness — they MUST use this function.
+
+    Semantics:
+    - listing.complete=True (authoritative):
+        Upserts all items, soft-deletes unseen children, marks folder FRESH,
+        advances last_refreshed_at.  Returns ListingResult(is_fresh=True).
+
+    - listing.complete=False (partial/unknown):
+        Upserts only seen items (never deletes unseen children), does NOT
+        advance last_refreshed_at, sets folder state to WARNING with stable
+        code 'incomplete_listing'.  Returns ListingResult(is_fresh=False, …).
+
+    No exception is raised on incomplete results — callers receive a structured
+    result and must propagate the warning state to the API response or task result.
+    This function never raises HTTPException — domain exceptions only.
+    This function never calls the quota service.
+    """
+    # Reconcile items from the provider listing
+    await reconcile_cloud_listing(
+        session,
+        user_id=user_id,
+        connection_id=connection_id,
+        parent_ref=parent_ref,
+        listing=listing,
+    )
+
+    if listing.complete:
+        # Authoritative: mark folder fresh and advance last_refreshed_at
+        await update_folder_state(
+            session,
+            user_id=user_id,
+            connection_id=connection_id,
+            parent_ref=parent_ref or "",
+            refresh_state="fresh",
+        )
+        return ListingResult(is_fresh=True)
+    else:
+        # Incomplete/partial: retain cached state; do NOT advance last_refreshed_at
+        # We only update the warning fields, preserving last_refreshed_at from prior success.
+        fs = await get_or_create_folder_state(
+            session,
+            user_id=user_id,
+            connection_id=connection_id,
+            parent_ref=parent_ref or "",
+        )
+        now = datetime.now(timezone.utc)
+        # Set warning state without touching last_refreshed_at
+        fs.refresh_state = "warning"
+        fs.error_code = "incomplete_listing"
+        fs.error_message = (
+            "Provider returned an incomplete listing. Showing cached results."
+        )
+        fs.updated_at = now
+        # Deliberately: do NOT update fs.last_refreshed_at
+        await session.flush()
+        return ListingResult(
+            is_fresh=False,
+            warning_code="incomplete_listing",
+            warning_message="Provider returned an incomplete listing. Showing cached results.",
+        )
+
+
 async def update_folder_state(
     session: AsyncSession,
     *,