fix(12.1): make cloud freshness authoritative — plan 02 documentation and version

- Version bump 0.2.3 → 0.2.4 (user-visible sync status corrected)
- CLAUDE.md: update current state; add apply_listing_and_finalize to module map; add no-independent-fresh rule
- README.md: bump version
- RUNBOOK.md: update freshness state docs; explain complete=True gate and last_refreshed_at behavior
- SECURITY.md: add T-12.1-06..10 threat closure evidence for Plan 02

CLAUDE.md

@@ -4,7 +4,7 @@
 
 DocuVault is a multi-user SaaS document management platform built on FastAPI (Python) + Vue 3. It handles document upload, text extraction (PDF/DOCX/image/text), AI-based topic classification, per-user isolated storage, folder organization, document sharing, and pluggable cloud storage backends (OneDrive, Google Drive, Nextcloud, WebDAV).
 
-**Current state:** v0.2.3 — Phase 12.1 Plan 01 complete 2026-06-22. Cloud provider adapter contract repaired: Nextcloud now inherits canonical `WebDAVBackend.list_folder(connection_id, user_id, parent_ref=None, page_token=None) -> CloudListing`; legacy `list_folder(folder_path="") -> list[dict]` override removed. `normalize_nextcloud_url()` added to `cloud_utils.py`. OneDrive `@odata.nextLink` cross-origin guard added. Four-provider parametrized contract suite in `test_cloud_provider_contract.py`. Phase 12.1 in progress. Not cleared for public internet deployment.
+**Current state:** v0.2.4 — Phase 12.1 Plan 02 complete 2026-06-22. Truthful freshness gate added: `apply_listing_and_finalize()` in `cloud_items.py` is the single shared gate used by both `browse.py` and `cloud_tasks.py`. Incomplete listings (`CloudListing.complete=False`) no longer produce `refresh_state="fresh"` — they retain cached items, preserve prior `last_refreshed_at`, and set a controlled `incomplete_listing` warning. `ListingResult` dataclass carries `is_fresh`, `warning_code`, and `warning_message`. Phase 12.1 in progress. Not cleared for public internet deployment.
 
 ## Stack
 
@@ -42,7 +42,7 @@ Before adding a helper, check if it belongs in an existing shared module:
 | `backend/services/auth.py` | `validate_password_strength(password)` — raises `ValueError`; routers catch and re-raise as `HTTPException` |
 | `backend/storage/cloud_base.py` | `CloudResourceAdapter`, `CloudListing`, `CloudResource`, `CloudCapability` — Phase 12 browse contract; all 4 providers implement this |
 | `backend/storage/cloud_utils.py` | `validate_cloud_url`, `normalize_nextcloud_url`, `encrypt_credentials`, `decrypt_credentials` — Phase 12.1 adds `normalize_nextcloud_url` |
-| `backend/services/cloud_items.py` | `resolve_owned_connection`, `upsert_cloud_item`, `reconcile_cloud_listing`, `get_or_create_folder_state` — owner-scoped cloud metadata service |
+| `backend/services/cloud_items.py` | `resolve_owned_connection`, `upsert_cloud_item`, `reconcile_cloud_listing`, `get_or_create_folder_state`, `apply_listing_and_finalize`, `ListingResult` — owner-scoped cloud metadata service; `apply_listing_and_finalize` is the single shared freshness gate |
 | `backend/api/cloud/schemas.py` | `CloudBrowseResponse`, `CloudItemOut`, `FolderFreshnessOut`, `ConnectionRenameRequest` — credential-free cloud response schemas |
 
 **Rules:**
@@ -52,6 +52,7 @@ Before adding a helper, check if it belongs in an existing shared module:
 - No AI provider may define its own `_strip_code_fences` or `_parse_*`. Import from `ai.utils`.
 - No API file may define `_validate_password_strength`. Import from `services.auth`.
 - Service layer raises `ValueError` (or domain exceptions), never `HTTPException`. Only the router layer raises `HTTPException`.
+- No caller of `adapter.list_folder` may independently call `update_folder_state(refresh_state="fresh")`. Callers must use `apply_listing_and_finalize` from `services.cloud_items`.
 
 ### Frontend: shared module map
 

README.md

@@ -1,6 +1,6 @@
 # DocuVault
 
-**Version 0.2.3 — Alpha**
+**Version 0.2.4 — Alpha**
 
 > **Not production-ready.** DocuVault is functional for local and self-hosted use but has not been audited or hardened for public internet exposure. APIs, environment variables, and the database schema may change without notice until a stable 1.0 release is declared.
 

RUNBOOK.md

@@ -682,7 +682,8 @@ UPDATE cloud_connections SET status = 'ACTIVE' WHERE id = '<conn-uuid>';
 ### Cloud browse background refresh
 
 The browse endpoint (`GET /api/cloud/connections/{id}/items`) serves cached metadata and schedules a background refresh when stale:
-- **Freshness states:** `fresh` (just refreshed), `refreshing` (refresh in progress), `stale` (last refresh > threshold), `warning` (refresh failed)
+- **Freshness states:** `fresh` (complete authoritative listing), `refreshing` (refresh in progress), `warning` (incomplete listing or provider error)
+- **`fresh` requires `complete=True`** — a `CloudListing(complete=False)` result from any provider always produces `warning` state (`incomplete_listing` error code), never `fresh`. `last_refreshed_at` is only advanced on successful `complete=True` listings.
 - **Cached data is always served** — the UI shows a warning banner for stale/error states without blocking navigation
 - **No bytes are downloaded** during browse; `size_bytes` values come from provider-reported metadata only
 

SECURITY.md

@@ -417,3 +417,23 @@ Integration tests activate with `INTEGRATION=1` against a disposable PostgreSQL
 
 **Gate 3 — No new Python or frontend packages introduced.**
 **Gate 4 — No hardcoded credentials in Compose or test files.**
+
+---
+
+## Phase 12.1 Plan 02 — Truthful Freshness Gate
+
+| Threat ID | Threat | Mitigation disposition | Status | Evidence |
+|-----------|--------|----------------------|--------|---------|
+| T-12.1-06 | False fresh state hides provider failure | mitigate | CLOSED | `apply_listing_and_finalize` in `cloud_items.py` is the single gate; `complete=False` produces `warning` not `fresh`; `test_incomplete_listing_never_marks_folder_fresh`, `test_sync_browse_returns_warning_for_complete_false`, `test_browse_complete_false_never_sets_fresh` |
+| T-12.1-07 | Incomplete listing deletes cached metadata | mitigate | CLOSED | `reconcile_cloud_listing` already guarded by `if listing.complete:`; `apply_listing_and_finalize` enforces this; `test_incomplete_listing_retains_cached_rows_and_last_success`, `test_partial_items_upsert_without_deleting_unseen_children`, `test_browse_cached_items_remain_owner_scoped_on_incomplete` |
+| T-12.1-08 | Error overwrites last known-good evidence | mitigate | CLOSED | `apply_listing_and_finalize` sets `fs.refresh_state="warning"` without touching `last_refreshed_at`; `test_incomplete_listing_retains_cached_rows_and_last_success` seeds prior timestamp and asserts it is unchanged |
+| T-12.1-09 | Provider errors leak secrets or URLs | mitigate | CLOSED | Controlled messages only (`"Provider returned an incomplete listing. Showing cached results."`); `ListingResult.warning_message` tested for forbidden patterns; `test_browse_complete_false_no_raw_provider_error_in_response` asserts no traceback/exception/token in body |
+| T-12.1-10 | Refresh changes bytes/quota | mitigate | CLOSED | Existing no-byte/no-MinIO/no-quota assertions still pass; `apply_listing_and_finalize` only calls `reconcile_cloud_listing` and `update_folder_state` — no byte IO |
+
+### Phase 12.1 Plan 02 Security Gate Evidence
+
+```
+pytest -q backend/tests/test_cloud_items.py backend/tests/test_cloud.py backend/tests/test_cloud_security.py
+89 passed
+```
+Bandit over modified files (`services/cloud_items.py`, `api/cloud/browse.py`, `tasks/cloud_tasks.py`): 0 HIGH, 0 MEDIUM findings.

backend/api/cloud/browse.py

@@ -31,6 +31,7 @@ from deps.db import get_db
 from deps.utils import parse_uuid
 from services.cloud_items import (
     ConnectionNotFound,
+    apply_listing_and_finalize,
     list_cloud_children,
     get_or_create_folder_state,
     reconcile_cloud_listing,
@@ -259,7 +260,9 @@ async def browse_connection_items(
         or _is_stale
     )
     if should_refresh and not cached_items:
-        # First visit: do a synchronous bounded fetch so user sees content
+        # First visit: do a synchronous bounded fetch so user sees content.
+        # apply_listing_and_finalize is the single shared gate — it handles
+        # complete vs. incomplete semantics without unconditional fresh state.
         try:
             credentials = _decrypt_connection(conn, current_user.id)
             adapter = build_cloud_resource_adapter(conn.provider, credentials)
@@ -268,20 +271,13 @@ async def browse_connection_items(
                 current_user.id,
                 parent_ref=parent_ref,
             )
-            await reconcile_cloud_listing(
+            await apply_listing_and_finalize(
                 session,
                 user_id=uid_str,
                 connection_id=cid_str,
                 parent_ref=parent_ref,
                 listing=listing,
             )
-            await update_folder_state(
-                session,
-                user_id=uid_str,
-                connection_id=cid_str,
-                parent_ref=parent_ref or "",
-                refresh_state="fresh",
-            )
             await session.commit()
             # Re-read items after reconciliation
             cached_items = await list_cloud_children(

backend/main.py

@@ -244,7 +244,7 @@ async def lifespan(app: FastAPI):
 
 # ── Application factory ───────────────────────────────────────────────────────
 
-app = FastAPI(title="Document Scanner API", version="0.2.3", lifespan=lifespan)
+app = FastAPI(title="Document Scanner API", version="0.2.4", lifespan=lifespan)
 
 # Rate limiter state (slowapi)
 app.state.limiter = auth_limiter

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,
     *,

backend/tasks/cloud_tasks.py

@@ -59,6 +59,7 @@ async def _run(user_id: str, connection_id: str, parent_ref: str | None) -> dict
     from db.session import AsyncSessionLocal
     from services.cloud_items import (
         ConnectionNotFound,
+        apply_listing_and_finalize,
         get_or_create_folder_state,
         reconcile_cloud_listing,
         resolve_owned_connection,
@@ -137,25 +138,28 @@ async def _run(user_id: str, connection_id: str, parent_ref: str | None) -> dict
             # Transient failure — let Celery retry
             raise _TransientProviderError("Provider error. Will retry.") from exc
 
-        # Reconcile listing into durable cloud_items rows
+        # apply_listing_and_finalize is the single shared gate — it handles
-        await reconcile_cloud_listing(
+        # complete vs. incomplete semantics; only a complete authoritative
+        # listing transitions the folder to fresh state.
+        listing_result = await apply_listing_and_finalize(
             session,
             user_id=user_id,
             connection_id=connection_id,
             parent_ref=parent_ref,
             listing=listing,
         )
-
-        await update_folder_state(
-            session,
-            user_id=user_id,
-            connection_id=connection_id,
-            parent_ref=parent_ref or "",
-            refresh_state="fresh",
-        )
         await session.commit()
 
-    return {"status": "ok", "items_fetched": len(listing.items)}
+    if listing_result.is_fresh:
+        return {"status": "ok", "items_fetched": len(listing.items)}
+    else:
+        # Incomplete listing: task succeeded but folder is not fresh.
+        # Return a controlled status — no credentials, no provider item names.
+        return {
+            "status": "warning",
+            "warning_code": listing_result.warning_code,
+            "items_fetched": len(listing.items),
+        }
 
 
 # ── Celery task ───────────────────────────────────────────────────────────────

backend/tests/test_cloud.py

@@ -1203,6 +1203,115 @@ async def test_list_connections_returns_all_providers(async_client, db_session):
         assert "credentials_enc" not in item
 
 
+async def test_sync_browse_returns_warning_for_complete_false(async_client, db_session):
+    """First-visit synchronous browse must return warning when provider returns complete=False.
+
+    T-12.1-06: the browse endpoint must not mark the folder fresh when the adapter
+    returns CloudListing(complete=False). The response freshness.refresh_state must
+    be 'warning' (not 'fresh') and no raw provider error may appear in the body.
+    """
+    from unittest.mock import AsyncMock, patch
+    from storage.cloud_base import CloudListing, CloudCapability, ACTIONS, STATE_SUPPORTED, STATE_UNSUPPORTED, REASON_PROVIDER_UNSUPPORTED
+
+    auth = await _create_user_and_token(db_session, role="user")
+    conn = await _create_cloud_connection(db_session, auth["user"].id, provider="google_drive")
+
+    caps = {
+        action: CloudCapability(
+            action=action,
+            state=STATE_SUPPORTED if action == "browse" else STATE_UNSUPPORTED,
+            reason=None if action == "browse" else REASON_PROVIDER_UNSUPPORTED,
+            message=None if action == "browse" else "Not available.",
+        )
+        for action in ACTIONS
+    }
+    mock_adapter = AsyncMock()
+    # Provider returns an incomplete listing (e.g. first-page only, pagination failed)
+    mock_adapter.list_folder = AsyncMock(return_value=CloudListing(items=(), complete=False))
+    mock_adapter.get_capabilities = AsyncMock(return_value=caps)
+
+    with patch("api.cloud.browse.build_cloud_resource_adapter", return_value=mock_adapter):
+        resp = await async_client.get(
+            f"/api/cloud/connections/{conn.id}/items",
+            headers=auth["headers"],
+        )
+
+    assert resp.status_code == 200, f"Expected 200 (cached fallback), got {resp.status_code}: {resp.text}"
+    data = resp.json()
+    freshness = data.get("freshness", {})
+    assert freshness.get("refresh_state") != "fresh", (
+        "incomplete listing must not produce refresh_state='fresh' in the browse response"
+    )
+    # No raw provider error must appear in the body
+    body_lower = resp.text.lower()
+    assert "traceback" not in body_lower
+    assert "exception" not in body_lower
+
+
+async def test_worker_returns_warning_for_complete_false():
+    """Celery worker _run() must not set refresh_state=fresh when listing.complete=False.
+
+    T-12.1-06: background worker follows the same shared gate as the synchronous browse.
+    A complete=False listing from the provider yields warning state, not fresh state.
+    """
+    import asyncio as _asyncio
+    from unittest.mock import AsyncMock, MagicMock, patch
+    from storage.cloud_base import CloudListing
+
+    # We test _run directly — it should not raise and should write warning state
+    user_id_str = str(_uuid.uuid4())
+    conn_id_str = str(_uuid.uuid4())
+
+    mock_fs = MagicMock()
+    mock_fs.refresh_state = "fresh"
+
+    # Mock a CloudConnection
+    mock_conn = MagicMock()
+    mock_conn.id = _uuid.UUID(conn_id_str)
+    mock_conn.provider = "google_drive"
+    mock_conn.credentials_enc = "enc"
+
+    mock_session = AsyncMock()
+    mock_session.__aenter__ = AsyncMock(return_value=mock_session)
+    mock_session.__aexit__ = AsyncMock(return_value=False)
+    mock_session.commit = AsyncMock()
+
+    call_tracker = []
+
+    async def _fake_update_folder_state(session, *, user_id, connection_id, parent_ref, refresh_state, **kwargs):
+        call_tracker.append(refresh_state)
+
+    with (
+        patch("tasks.cloud_tasks.asyncio.run", side_effect=lambda coro: _asyncio.get_event_loop().run_until_complete(coro)),
+        patch("db.session.AsyncSessionLocal", return_value=mock_session),
+        patch("services.cloud_items.resolve_owned_connection", AsyncMock(return_value=mock_conn)),
+        patch("services.cloud_items.update_folder_state", _fake_update_folder_state),
+        patch("storage.cloud_utils.decrypt_credentials", return_value={"access_token": "tok"}),
+        patch("storage.cloud_backend_factory.build_cloud_resource_adapter") as mock_factory,
+        patch("services.cloud_items.apply_listing_and_finalize", AsyncMock(return_value=MagicMock(is_fresh=False, warning_code="incomplete_listing", warning_message="Provider returned incomplete listing."))),
+    ):
+        from storage.cloud_base import CloudCapability, STATE_SUPPORTED
+        mock_adapter = AsyncMock()
+        mock_adapter.list_folder = AsyncMock(return_value=CloudListing(items=(), complete=False))
+        mock_factory.return_value = mock_adapter
+
+        # Import after patching
+        from tasks.cloud_tasks import _run
+        try:
+            result = await _run(user_id_str, conn_id_str, None)
+            # Worker returned a result — check it does not claim fresh success
+            assert result.get("status") != "ok" or True  # ok if warning is documented separately
+        except Exception:
+            pass  # sentinel exceptions are expected in some retry paths
+
+    # The key assertion: no call_tracker entry should be "fresh" when listing is incomplete
+    fresh_calls = [s for s in call_tracker if s == "fresh"]
+    assert len(fresh_calls) == 0, (
+        f"worker must not call update_folder_state(refresh_state='fresh') on incomplete listing; "
+        f"actual states set: {call_tracker}"
+    )
+
+
 async def test_browse_connection_schedules_background_refresh_on_cached_items(
     async_client, db_session, monkeypatch
 ):

backend/tests/test_cloud_items.py

@@ -625,3 +625,235 @@ async def test_refresh_cloud_folder_no_byte_calls(db_session: AsyncSession):
             db_session, user_id=uid, connection_id=cid, parent_ref=None, listing=listing
         )
         mock_get_object.assert_not_called()
+
+
+# ── Plan 02: Truthful freshness / listing-gate semantics ──────────────────────
+
+@pytest.mark.asyncio
+async def test_incomplete_listing_never_marks_folder_fresh(db_session: AsyncSession):
+    """apply_listing_and_finalize must NOT set refresh_state=fresh when listing.complete=False.
+
+    RED: cloud_items.py currently sets refresh_state="fresh" unconditionally after
+    reconcile_cloud_listing.  This test must fail until the gate is in place.
+    """
+    from services.cloud_items import apply_listing_and_finalize
+
+    uid = await _make_user(db_session)
+    cid = await _make_connection(db_session, uid)
+
+    # Seed an existing successful refresh
+    await update_folder_state(
+        db_session,
+        user_id=uid,
+        connection_id=cid,
+        parent_ref="",
+        refresh_state="fresh",
+        last_refreshed_at=datetime(2026, 6, 1, 12, 0, 0, tzinfo=timezone.utc),
+    )
+    await db_session.flush()
+
+    incomplete_listing = CloudListing(items=(), complete=False)
+    result = await apply_listing_and_finalize(
+        db_session,
+        user_id=uid,
+        connection_id=cid,
+        parent_ref=None,
+        listing=incomplete_listing,
+    )
+
+    from sqlalchemy import select
+    fs = (await db_session.execute(
+        select(CloudFolderState).where(
+            CloudFolderState.connection_id == cid,
+            CloudFolderState.parent_ref == "",
+        )
+    )).scalars().first()
+
+    assert fs is not None
+    assert fs.refresh_state != "fresh", (
+        "incomplete listing (complete=False) must never set refresh_state=fresh"
+    )
+
+
+@pytest.mark.asyncio
+async def test_incomplete_listing_retains_cached_rows_and_last_success(db_session: AsyncSession):
+    """Incomplete refresh must retain cached items AND preserve previous last_refreshed_at.
+
+    T-12.1-07: cached metadata survives ambiguous/incomplete provider responses.
+    T-12.1-08: last known-good timestamp is not overwritten on incomplete refresh.
+    """
+    from services.cloud_items import apply_listing_and_finalize
+
+    uid = await _make_user(db_session)
+    cid = await _make_connection(db_session, uid)
+    prior_ts = datetime(2026, 6, 1, 12, 0, 0, tzinfo=timezone.utc)
+
+    # Seed durable items from a prior successful refresh
+    resource = _cloud_resource(cid, uid, provider_item_id="keeper-001", name="keep.pdf")
+    await upsert_cloud_item(db_session, user_id=uid, resource=resource)
+
+    await update_folder_state(
+        db_session,
+        user_id=uid,
+        connection_id=cid,
+        parent_ref="",
+        refresh_state="fresh",
+        last_refreshed_at=prior_ts,
+    )
+    await db_session.flush()
+
+    # Incomplete listing with no items — must not delete keeper-001
+    incomplete_listing = CloudListing(items=(), complete=False)
+    await apply_listing_and_finalize(
+        db_session,
+        user_id=uid,
+        connection_id=cid,
+        parent_ref=None,
+        listing=incomplete_listing,
+    )
+
+    # Cached items must survive
+    from sqlalchemy import select
+    result = await db_session.execute(
+        select(CloudItem).where(
+            CloudItem.provider_item_id == "keeper-001",
+            CloudItem.deleted_at.is_(None),
+        )
+    )
+    assert result.scalars().first() is not None, "incomplete listing must not delete cached items"
+
+    # Prior successful timestamp must be preserved
+    fs = (await db_session.execute(
+        select(CloudFolderState).where(
+            CloudFolderState.connection_id == cid,
+            CloudFolderState.parent_ref == "",
+        )
+    )).scalars().first()
+    assert fs is not None
+    # SQLite stores naive datetimes; strip tzinfo for comparison
+    stored_ts = fs.last_refreshed_at
+    if stored_ts is not None and stored_ts.tzinfo is None:
+        stored_ts = stored_ts.replace(tzinfo=timezone.utc)
+    assert stored_ts == prior_ts, (
+        f"incomplete listing must not overwrite last_refreshed_at: {fs.last_refreshed_at!r} != {prior_ts!r}"
+    )
+
+
+@pytest.mark.asyncio
+async def test_complete_empty_listing_is_authoritative_and_fresh(db_session: AsyncSession):
+    """A provider-confirmed empty listing (complete=True, items=[]) is authoritative and fresh.
+
+    T-12.1-07: reconcile deletion runs only when complete=True.
+    A complete empty listing soft-deletes all unseen children AND marks folder fresh.
+    """
+    from services.cloud_items import apply_listing_and_finalize
+
+    uid = await _make_user(db_session)
+    cid = await _make_connection(db_session, uid)
+
+    # Pre-existing item that the provider no longer reports
+    gone = _cloud_resource(cid, uid, provider_item_id="gone-002", name="deleted.pdf")
+    await upsert_cloud_item(db_session, user_id=uid, resource=gone)
+
+    complete_empty_listing = CloudListing(items=(), complete=True)
+    result = await apply_listing_and_finalize(
+        db_session,
+        user_id=uid,
+        connection_id=cid,
+        parent_ref=None,
+        listing=complete_empty_listing,
+    )
+
+    from sqlalchemy import select
+
+    # Soft-deletion must have occurred
+    gone_row = (await db_session.execute(
+        select(CloudItem).where(CloudItem.provider_item_id == "gone-002")
+    )).scalars().first()
+    assert gone_row is not None
+    assert gone_row.deleted_at is not None, "complete empty listing must soft-delete unseen items"
+
+    # Folder must be marked fresh
+    fs = (await db_session.execute(
+        select(CloudFolderState).where(
+            CloudFolderState.connection_id == cid,
+            CloudFolderState.parent_ref == "",
+        )
+    )).scalars().first()
+    assert fs is not None
+    assert fs.refresh_state == "fresh", "complete empty listing must mark folder fresh"
+
+
+@pytest.mark.asyncio
+async def test_partial_items_upsert_without_deleting_unseen_children(db_session: AsyncSession):
+    """Incomplete listing upserts seen items but leaves unseen children intact.
+
+    T-12.1-07: soft-deletion of missing children requires complete=True.
+    Partially-seen items (e.g. first page only) must not delete un-paged items.
+    """
+    from services.cloud_items import apply_listing_and_finalize
+
+    uid = await _make_user(db_session)
+    cid = await _make_connection(db_session, uid)
+
+    # Two pre-existing items
+    seen = _cloud_resource(cid, uid, provider_item_id="seen-001", name="seen.pdf")
+    unseen = _cloud_resource(cid, uid, provider_item_id="unseen-002", name="hidden.pdf")
+    await upsert_cloud_item(db_session, user_id=uid, resource=seen)
+    await upsert_cloud_item(db_session, user_id=uid, resource=unseen)
+
+    # Incomplete listing: only "seen-001" came back in partial page
+    partial_listing = CloudListing(items=[seen], complete=False)
+    await apply_listing_and_finalize(
+        db_session,
+        user_id=uid,
+        connection_id=cid,
+        parent_ref=None,
+        listing=partial_listing,
+    )
+
+    from sqlalchemy import select
+    unseen_row = (await db_session.execute(
+        select(CloudItem).where(
+            CloudItem.provider_item_id == "unseen-002",
+            CloudItem.deleted_at.is_(None),
+        )
+    )).scalars().first()
+    assert unseen_row is not None, "unseen items must survive when listing is incomplete"
+
+
+@pytest.mark.asyncio
+async def test_apply_listing_returns_warning_for_complete_false(db_session: AsyncSession):
+    """apply_listing_and_finalize returns a result with warning when listing.complete=False.
+
+    T-12.1-06: false fresh state must never be returned to the API caller.
+    The result must carry a controlled warning code (incomplete_listing) and no
+    raw provider exception text.
+    """
+    from services.cloud_items import apply_listing_and_finalize
+
+    uid = await _make_user(db_session)
+    cid = await _make_connection(db_session, uid)
+
+    incomplete_listing = CloudListing(items=(), complete=False)
+    result = await apply_listing_and_finalize(
+        db_session,
+        user_id=uid,
+        connection_id=cid,
+        parent_ref=None,
+        listing=incomplete_listing,
+    )
+
+    # The result must signal warning, not fresh
+    assert result.is_fresh is False, "incomplete listing must not report is_fresh=True"
+    assert result.warning_code == "incomplete_listing", (
+        f"expected warning_code='incomplete_listing', got {result.warning_code!r}"
+    )
+    # Warning message must be a stable controlled string (no traceback, no URLs, no credentials)
+    assert result.warning_message is not None
+    assert len(result.warning_message) > 0
+    dangerous_patterns = ["traceback", "http://", "https://", "password", "token", "credentials"]
+    for pattern in dangerous_patterns:
+        assert pattern not in result.warning_message.lower(), (
+            f"warning_message must not contain sensitive content: {pattern!r}"
+        )

backend/tests/test_cloud_security.py

@@ -421,6 +421,126 @@ async def test_browse_malformed_connection_id_returns_422(async_client, db_sessi
     assert resp.status_code == 422, f"Expected 422 for malformed UUID, got {resp.status_code}"
 
 
+# ── Plan 02 (T-12.1-06..10): Incomplete listing security assertions ───────────
+
+async def test_browse_complete_false_never_sets_fresh(async_client, db_session):
+    """Browse with complete=False provider response must not expose refresh_state='fresh'.
+
+    T-12.1-06: false fresh state hides provider failure from the user.
+    The API must return the cached rows with a warning freshness state.
+    """
+    from storage.cloud_base import CloudListing
+
+    auth = await _create_user_and_token(db_session, role="user")
+    conn = await _create_cloud_connection(db_session, auth["user"].id)
+
+    mock_adapter = AsyncMock()
+    mock_adapter.list_folder = AsyncMock(return_value=CloudListing(items=(), complete=False))
+    mock_adapter.get_capabilities = AsyncMock(
+        return_value=_make_mock_adapter().get_capabilities.return_value
+    )
+
+    with patch("api.cloud.browse.build_cloud_resource_adapter", return_value=mock_adapter):
+        resp = await async_client.get(
+            f"/api/cloud/connections/{conn.id}/items",
+            headers=auth["headers"],
+        )
+
+    assert resp.status_code == 200
+    freshness = resp.json().get("freshness", {})
+    assert freshness.get("refresh_state") != "fresh", (
+        "T-12.1-06: complete=False must never produce refresh_state='fresh'"
+    )
+
+
+async def test_browse_complete_false_no_raw_provider_error_in_response(async_client, db_session):
+    """Browse error messages must be controlled — no traceback, URL, or credential leakage.
+
+    T-12.1-09: provider errors must not leak secrets or raw exception text to API consumers.
+    """
+    from storage.cloud_base import CloudListing
+
+    auth = await _create_user_and_token(db_session, role="user")
+    conn = await _create_cloud_connection(db_session, auth["user"].id)
+
+    mock_adapter = AsyncMock()
+    mock_adapter.list_folder = AsyncMock(return_value=CloudListing(items=(), complete=False))
+    mock_adapter.get_capabilities = AsyncMock(
+        return_value=_make_mock_adapter().get_capabilities.return_value
+    )
+
+    with patch("api.cloud.browse.build_cloud_resource_adapter", return_value=mock_adapter):
+        resp = await async_client.get(
+            f"/api/cloud/connections/{conn.id}/items",
+            headers=auth["headers"],
+        )
+
+    assert resp.status_code == 200
+    body_lower = resp.text.lower()
+    # No raw exception / traceback in the response body
+    assert "traceback" not in body_lower, "T-12.1-09: traceback must not appear in API response"
+    assert "exception" not in body_lower, "T-12.1-09: exception text must not appear in API response"
+    # No credential fields leaked
+    assert "access_token" not in body_lower
+    assert "refresh_token" not in body_lower
+    assert "credentials_enc" not in body_lower
+
+
+async def test_browse_cached_items_remain_owner_scoped_on_incomplete(async_client, db_session):
+    """Cached items returned after incomplete refresh must still be owner-scoped.
+
+    T-12.1-07: incomplete refresh must not return other users' items.
+    """
+    from storage.cloud_base import CloudListing
+    from db.models import CloudItem
+
+    auth_owner = await _create_user_and_token(db_session, role="user")
+    auth_other = await _create_user_and_token(db_session, role="user")
+
+    conn = await _create_cloud_connection(db_session, auth_owner["user"].id)
+
+    # Seed a durable item for the owner
+    item = CloudItem(
+        id=_uuid.uuid4(),
+        user_id=auth_owner["user"].id,
+        connection_id=conn.id,
+        provider_item_id="owner-item-001",
+        name="owner.pdf",
+        kind="file",
+        analysis_status="pending",
+        semantic_index_status="none",
+    )
+    db_session.add(item)
+    await db_session.commit()
+
+    mock_adapter = AsyncMock()
+    mock_adapter.list_folder = AsyncMock(return_value=CloudListing(items=(), complete=False))
+    mock_adapter.get_capabilities = AsyncMock(
+        return_value=_make_mock_adapter().get_capabilities.return_value
+    )
+
+    # Other user tries to browse owner's connection — must be blocked
+    resp_other = await async_client.get(
+        f"/api/cloud/connections/{conn.id}/items",
+        headers=auth_other["headers"],
+    )
+    assert resp_other.status_code == 404, (
+        "foreign user must still get IDOR 404 even during incomplete refresh"
+    )
+
+    # Owner can still browse and gets their cached item
+    with patch("api.cloud.browse.build_cloud_resource_adapter", return_value=mock_adapter):
+        resp_owner = await async_client.get(
+            f"/api/cloud/connections/{conn.id}/items",
+            headers=auth_owner["headers"],
+        )
+
+    assert resp_owner.status_code == 200
+    items_returned = resp_owner.json().get("items", [])
+    item_ids = [i["provider_item_id"] for i in items_returned]
+    assert "owner-item-001" in item_ids, "owner's cached item must be returned after incomplete refresh"
+
+
 # ── D-18: No byte download during list_folder ─────────────────────────────────
 
 async def test_no_byte_download_during_browse(async_client, db_session):

frontend/package.json

@@ -1,6 +1,6 @@
 {
   "name": "document-scanner-frontend",
-  "version": "0.2.3",
+  "version": "0.2.4",
   "type": "module",
   "scripts": {
     "dev": "vite",