test(13-06): add failing upload reconciliation tests (RED)

- test_upload_success_upserts_cloud_item_before_returning — verifies CloudItem row exists after success
- test_upload_success_marks_folder_freshness_stale — verifies folder state updated on success
- test_upload_failed_does_not_mutate_cloud_items — verifies failed uploads don't create phantom items

backend/tests/test_cloud_mutations.py

@@ -1045,3 +1045,187 @@ async def test_upload_foreign_user_blocked(async_client, db_session):
     assert resp.status_code == 404, (
         f"Expected 404 IDOR block for foreign upload, got {resp.status_code}"
     )
+
+
+# ── Phase 13 Plan 06 Task 1 (RED): Upload success routes through reconciliation ─
+
+
+async def test_upload_success_upserts_cloud_item_before_returning(async_client, db_session):
+    """POST upload success upserts the uploaded item into cloud_items before returning.
+
+    Plan 06 Task 1 behavior 1: Successful upload must update navigation metadata
+    through centralized reconciliation (cloud_items.upsert_cloud_item) before the
+    route returns success — not as a fire-and-forget side effect.
+
+    The returned 'uploaded' body's provider_item_id must have a matching CloudItem
+    row in the DB by the time the response is received.
+
+    FAILS: Upload route does not call upsert_cloud_item on success yet.
+    """
+    from storage.cloud_base import MUT_KIND_UPLOADED, MUT_REASON_CREATED
+    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)
+
+    new_provider_item_id = str(_uuid.uuid4())
+
+    import uuid as _uuid2
+    mock_adapter = _make_mock_mutable_adapter(
+        upload_result={
+            "kind": MUT_KIND_UPLOADED,
+            "reason": MUT_REASON_CREATED,
+            "provider_item_id": new_provider_item_id,
+            "name": "reconcile_test.pdf",
+            "parent_ref": "reconcile_folder",
+            "size": 2048,
+        }
+    )
+
+    files = {"file": ("reconcile_test.pdf", b"%PDF-1.4 reconcile", "application/pdf")}
+    data = {"parent_ref": "reconcile_folder_unique", "filename": "reconcile_test.pdf"}
+
+    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/upload",
+            headers=auth["headers"],
+            files=files,
+            data=data,
+        )
+
+    assert resp.status_code == 200, f"Expected 200 for upload success, got {resp.status_code}"
+    body = resp.json()
+    assert body.get("kind") == "uploaded"
+
+    # After successful upload, the provider_item_id must have a CloudItem row
+    result = await db_session.execute(
+        select(CloudItem).where(
+            CloudItem.connection_id == conn.id,
+            CloudItem.provider_item_id == new_provider_item_id,
+        )
+    )
+    cloud_item = result.scalar_one_or_none()
+    assert cloud_item is not None, (
+        "Upload success must upsert a CloudItem row before returning (Plan 06 Task 1 behavior 1)"
+    )
+    assert cloud_item.name == "reconcile_test.pdf", (
+        f"Upserted CloudItem must have the correct name, got {cloud_item.name!r}"
+    )
+    assert cloud_item.kind == "file", (
+        f"Upserted CloudItem must have kind='file', got {cloud_item.kind!r}"
+    )
+
+
+async def test_upload_success_marks_folder_freshness_stale(async_client, db_session):
+    """POST upload success marks the parent folder freshness as stale (not fresh).
+
+    Plan 06 Task 1 behavior 2: After a successful upload, the parent folder listing
+    has changed — the folder state must be invalidated (set to 'warning' or 'stale')
+    so the next browse triggers a provider re-list. Setting it 'fresh' with an old
+    listing would be a lie.
+
+    The implementation must NOT call apply_listing_and_finalize (which requires a full
+    provider listing). It must call update_folder_state with a non-'fresh' state that
+    signals the folder needs refreshing.
+
+    FAILS: Upload route does not update folder freshness state yet.
+    """
+    from storage.cloud_base import MUT_KIND_UPLOADED, MUT_REASON_CREATED
+    from db.models import CloudFolderState
+    from sqlalchemy import select
+
+    auth = await _create_user_and_token(db_session)
+    conn = await _create_cloud_connection(db_session, auth["user"].id)
+
+    new_provider_item_id = str(_uuid.uuid4())
+    test_parent_ref = "freshness_test_folder"
+
+    mock_adapter = _make_mock_mutable_adapter(
+        upload_result={
+            "kind": MUT_KIND_UPLOADED,
+            "reason": MUT_REASON_CREATED,
+            "provider_item_id": new_provider_item_id,
+            "name": "freshness_test.pdf",
+            "parent_ref": test_parent_ref,
+            "size": 512,
+        }
+    )
+
+    files = {"file": ("freshness_test.pdf", b"content", "application/pdf")}
+    data = {"parent_ref": test_parent_ref, "filename": "freshness_test.pdf"}
+
+    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/upload",
+            headers=auth["headers"],
+            files=files,
+            data=data,
+        )
+
+    assert resp.status_code == 200
+    assert resp.json().get("kind") == "uploaded"
+
+    # Parent folder state must be refreshed/invalidated — not left as stale from before
+    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()
+    # After upload, a folder state row must exist to signal refresh needed
+    assert fs is not None, (
+        "Upload success must create/update a CloudFolderState row for the parent folder "
+        "(Plan 06 Task 1 behavior 2)"
+    )
+
+
+async def test_upload_failed_does_not_mutate_cloud_items(async_client, db_session):
+    """Failed, skipped, and canceled queue decisions do not create CloudItem rows.
+
+    Plan 06 Task 1 behavior 3: Only authoritative upload success may mutate listing
+    state. An 'offline' or 'reauth_required' upload result must not upsert a CloudItem
+    row — that would corrupt the navigation cache with phantom items.
+
+    FAILS: Current behavior is correct (no reconcile call at all), but this test
+    establishes the invariant explicitly for regression.
+    """
+    from storage.cloud_base import MUT_KIND_OFFLINE, MUT_REASON_PROVIDER_OFFLINE
+    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)
+
+    before_count_result = await db_session.execute(
+        select(CloudItem).where(CloudItem.connection_id == conn.id)
+    )
+    before_count = len(before_count_result.scalars().all())
+
+    mock_adapter = _make_mock_mutable_adapter(
+        upload_result={"kind": MUT_KIND_OFFLINE, "reason": MUT_REASON_PROVIDER_OFFLINE}
+    )
+
+    files = {"file": ("offline_test.pdf", b"data", "application/pdf")}
+    data = {"parent_ref": "some_folder", "filename": "offline_phantom.pdf"}
+
+    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/upload",
+            headers=auth["headers"],
+            files=files,
+            data=data,
+        )
+
+    assert resp.status_code != 500
+
+    # No new CloudItem must have been created for the failed upload
+    after_result = await db_session.execute(
+        select(CloudItem).where(CloudItem.connection_id == conn.id)
+    )
+    after_count = len(after_result.scalars().all())
+    assert after_count == before_count, (
+        f"Failed upload must not create CloudItem rows (Plan 06 Task 1 behavior 3): "
+        f"count changed {before_count} → {after_count}"
+    )