feat(12-01): durable owner-scoped cloud metadata schema (migration 0006 + models)

- Migration 0006: cloud_items, cloud_item_topics, cloud_folder_states tables
- cloud_connections: add display_name_override column for same-provider disambiguation
- CloudItem, CloudItemTopic, CloudFolderState ORM models with owner/connection indexes
- Unique (connection_id, provider_item_id) boundary; no MinIO object_key field
- Root folder state representable as parent_ref='' without CloudItem parent row
- services/cloud_items.py: resolve_owned_connection, upsert, list, reconcile, folder state
- 17 unit/integration tests covering model fields, isolation, quota invariant, idempotency

backend/db/models.py

@@ -318,6 +318,144 @@ class CloudConnection(Base):
     __table_args__ = (Index("ix_cloud_connections_user", "user_id"),)
 
 
+class CloudItem(Base):
+    """Durable per-item cloud metadata indexed by (connection_id, provider_item_id).
+
+    Phase 12: metadata browsing only — no MinIO object_key field.
+    Phase 14: extracted_text, analysis_status, semantic_index_status, semantic_index_data
+              are reserved for byte analysis and semantic search.
+
+    D-18: provider_size never flows to quotas.used_bytes.
+    """
+
+    __tablename__ = "cloud_items"
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+    user_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("users.id", ondelete="CASCADE"),
+        nullable=False,
+    )
+    connection_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("cloud_connections.id", ondelete="CASCADE"),
+        nullable=False,
+    )
+    provider_item_id: Mapped[str] = mapped_column(Text, nullable=False)
+    parent_ref: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
+    path_snapshot: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
+    name: Mapped[str] = mapped_column(Text, nullable=False)
+    kind: Mapped[str] = mapped_column(String(8), nullable=False)  # "file" | "folder"
+    content_type: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
+    # Provider-reported size — never used to update quotas.used_bytes (D-18)
+    provider_size: Mapped[Optional[int]] = mapped_column(BigInteger, nullable=True)
+    etag: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
+    version: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
+    modified_at: Mapped[Optional[datetime]] = mapped_column(
+        TIMESTAMP(timezone=True), nullable=True
+    )
+    last_seen_at: Mapped[datetime] = mapped_column(
+        TIMESTAMP(timezone=True), nullable=False, server_default=func.now()
+    )
+    deleted_at: Mapped[Optional[datetime]] = mapped_column(
+        TIMESTAMP(timezone=True), nullable=True
+    )
+    # Phase 14 reserved fields
+    extracted_text: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
+    analysis_status: Mapped[str] = mapped_column(
+        String(16), nullable=False, default="pending"
+    )
+    semantic_index_status: Mapped[str] = mapped_column(
+        String(16), nullable=False, default="none"
+    )
+    semantic_index_data: Mapped[Optional[dict]] = mapped_column(JSONB, nullable=True)
+    created_at: Mapped[datetime] = mapped_column(
+        TIMESTAMP(timezone=True), nullable=False, server_default=func.now()
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        TIMESTAMP(timezone=True), nullable=False, server_default=func.now()
+    )
+
+    __table_args__ = (
+        UniqueConstraint(
+            "connection_id",
+            "provider_item_id",
+            name="uq_cloud_items_connection_provider_item",
+        ),
+        Index("ix_cloud_items_user_id", "user_id"),
+        Index("ix_cloud_items_connection_id", "connection_id"),
+        Index("ix_cloud_items_connection_parent", "connection_id", "parent_ref"),
+    )
+
+
+class CloudItemTopic(Base):
+    """Association between CloudItem and Topic — no Document row required."""
+
+    __tablename__ = "cloud_item_topics"
+
+    cloud_item_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("cloud_items.id", ondelete="CASCADE"),
+        primary_key=True,
+    )
+    topic_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("topics.id", ondelete="CASCADE"),
+        primary_key=True,
+    )
+
+
+class CloudFolderState(Base):
+    """Per-connection folder freshness row for cached-first navigation.
+
+    parent_ref = '' represents the connection root (allows unique constraint).
+    refresh_state: "refreshing" | "fresh" | "warning"
+    """
+
+    __tablename__ = "cloud_folder_states"
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+    user_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("users.id", ondelete="CASCADE"),
+        nullable=False,
+    )
+    connection_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("cloud_connections.id", ondelete="CASCADE"),
+        nullable=False,
+    )
+    # '' for root, provider ref for non-root
+    parent_ref: Mapped[str] = mapped_column(Text, nullable=False, default="")
+    refresh_state: Mapped[str] = mapped_column(
+        String(16), nullable=False, default="fresh"
+    )
+    last_refreshed_at: Mapped[Optional[datetime]] = mapped_column(
+        TIMESTAMP(timezone=True), nullable=True
+    )
+    error_code: Mapped[Optional[str]] = mapped_column(String(64), nullable=True)
+    error_message: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
+    created_at: Mapped[datetime] = mapped_column(
+        TIMESTAMP(timezone=True), nullable=False, server_default=func.now()
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        TIMESTAMP(timezone=True), nullable=False, server_default=func.now()
+    )
+
+    __table_args__ = (
+        UniqueConstraint(
+            "connection_id",
+            "parent_ref",
+            name="uq_cloud_folder_states_connection_parent",
+        ),
+        Index("ix_cloud_folder_states_connection", "connection_id"),
+    )
+
+
 class Group(Base):
     """v2 stub — empty table, seeded for schema completeness (PROJECT.md D-02).
 

backend/migrations/versions/0006_cloud_resource_foundation.py

@@ -0,0 +1,208 @@
+"""Add cloud_items, cloud_item_topics, and cloud_folder_states tables.
+
+Revision ID: 0006
+Revises: 0005
+Create Date: 2026-06-18
+
+Changes:
+  1. cloud_connections: add display_name_override (nullable Text) for user-defined
+     display names distinct from the auto-generated default.  Does NOT add a
+     (user_id, provider) unique constraint — multiple same-provider accounts must
+     remain permitted (D-02, D-03).
+
+  2. cloud_items: durable per-item metadata indexed by connection/provider item ID.
+     No MinIO object_key field — provider bytes are never mirrored here.
+     analysis_status, semantic_index_status, semantic_index_data reserved for
+     Phase 14 analysis and semantic search.
+
+  3. cloud_item_topics: association between cloud items and existing owner topics.
+     No local Document row required.
+
+  4. cloud_folder_states: per-connection/parent-ref folder freshness row.
+     Root folder represented with parent_ref = '' (empty string, not NULL) to
+     allow a unique constraint on (connection_id, parent_ref).
+
+Design notes:
+  - Ownership boundary: every cloud row has user_id + connection_id.
+  - Uniqueness: (connection_id, provider_item_id) for items; (connection_id, parent_ref)
+    for folder states.  Same provider_item_id can coexist across connections.
+  - CASCADE: deleting a user or connection cascades all child rows automatically.
+  - D-18 compliance: no quota_used / byte_size field that flows to quotas.used_bytes.
+"""
+from __future__ import annotations
+
+import sqlalchemy as sa
+from sqlalchemy.dialects.postgresql import UUID as PG_UUID, JSONB
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "0006"
+down_revision = "0005"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # 1. Extend cloud_connections with optional user-defined display name override
+    op.add_column(
+        "cloud_connections",
+        sa.Column("display_name_override", sa.Text, nullable=True),
+    )
+
+    # 2. cloud_items table
+    op.create_table(
+        "cloud_items",
+        sa.Column(
+            "id",
+            PG_UUID(as_uuid=True),
+            primary_key=True,
+            server_default=sa.text("gen_random_uuid()"),
+        ),
+        sa.Column(
+            "user_id",
+            PG_UUID(as_uuid=True),
+            sa.ForeignKey("users.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column(
+            "connection_id",
+            PG_UUID(as_uuid=True),
+            sa.ForeignKey("cloud_connections.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        # Opaque provider-assigned identifier (e.g. Drive file ID, OneDrive driveItem id)
+        sa.Column("provider_item_id", sa.Text, nullable=False),
+        # Opaque reference to the parent folder (empty string = root)
+        sa.Column("parent_ref", sa.Text, nullable=True),
+        # Provider-reported path snapshot — informational only, not used for lookup
+        sa.Column("path_snapshot", sa.Text, nullable=True),
+        sa.Column("name", sa.Text, nullable=False),
+        # "file" | "folder"
+        sa.Column("kind", sa.String(8), nullable=False),
+        sa.Column("content_type", sa.Text, nullable=True),
+        # Provider-reported size — never flows to quotas.used_bytes (D-18)
+        sa.Column("provider_size", sa.BigInteger, nullable=True),
+        sa.Column("etag", sa.Text, nullable=True),
+        sa.Column("version", sa.Text, nullable=True),
+        sa.Column("modified_at", sa.TIMESTAMP(timezone=True), nullable=True),
+        sa.Column(
+            "last_seen_at",
+            sa.TIMESTAMP(timezone=True),
+            nullable=False,
+            server_default=sa.text("now()"),
+        ),
+        sa.Column("deleted_at", sa.TIMESTAMP(timezone=True), nullable=True),
+        # Phase 14: byte analysis fields (reserved)
+        sa.Column("extracted_text", sa.Text, nullable=True),
+        # "pending" | "processing" | "done" | "error" | "skipped"
+        sa.Column("analysis_status", sa.String(16), nullable=False, server_default="pending"),
+        # "none" | "pending" | "indexed" | "error"
+        sa.Column("semantic_index_status", sa.String(16), nullable=False, server_default="none"),
+        # Provider-independent semantic search artifacts (reserved for Phase 15)
+        sa.Column("semantic_index_data", JSONB, nullable=True),
+        sa.Column(
+            "created_at",
+            sa.TIMESTAMP(timezone=True),
+            nullable=False,
+            server_default=sa.text("now()"),
+        ),
+        sa.Column(
+            "updated_at",
+            sa.TIMESTAMP(timezone=True),
+            nullable=False,
+            server_default=sa.text("now()"),
+        ),
+        # Uniqueness: same provider item ID can exist in different connections only
+        sa.UniqueConstraint(
+            "connection_id", "provider_item_id",
+            name="uq_cloud_items_connection_provider_item",
+        ),
+    )
+
+    # Indexes for ownership lookups
+    op.create_index("ix_cloud_items_user_id", "cloud_items", ["user_id"])
+    op.create_index("ix_cloud_items_connection_id", "cloud_items", ["connection_id"])
+    op.create_index(
+        "ix_cloud_items_connection_parent",
+        "cloud_items",
+        ["connection_id", "parent_ref"],
+    )
+
+    # 3. cloud_item_topics association table
+    op.create_table(
+        "cloud_item_topics",
+        sa.Column(
+            "cloud_item_id",
+            PG_UUID(as_uuid=True),
+            sa.ForeignKey("cloud_items.id", ondelete="CASCADE"),
+            primary_key=True,
+        ),
+        sa.Column(
+            "topic_id",
+            PG_UUID(as_uuid=True),
+            sa.ForeignKey("topics.id", ondelete="CASCADE"),
+            primary_key=True,
+        ),
+    )
+
+    # 4. cloud_folder_states table
+    # parent_ref uses empty string '' for root (allows unique constraint)
+    op.create_table(
+        "cloud_folder_states",
+        sa.Column(
+            "id",
+            PG_UUID(as_uuid=True),
+            primary_key=True,
+            server_default=sa.text("gen_random_uuid()"),
+        ),
+        sa.Column(
+            "user_id",
+            PG_UUID(as_uuid=True),
+            sa.ForeignKey("users.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column(
+            "connection_id",
+            PG_UUID(as_uuid=True),
+            sa.ForeignKey("cloud_connections.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        # '' for root folder, provider ref string for non-root
+        sa.Column("parent_ref", sa.Text, nullable=False, server_default=""),
+        # "refreshing" | "fresh" | "warning"
+        sa.Column("refresh_state", sa.String(16), nullable=False, server_default="fresh"),
+        sa.Column("last_refreshed_at", sa.TIMESTAMP(timezone=True), nullable=True),
+        # Stable error code (e.g. "token_expired", "rate_limited") — never raw exception
+        sa.Column("error_code", sa.String(64), nullable=True),
+        # Safe human-readable message — never raw provider error text
+        sa.Column("error_message", sa.Text, nullable=True),
+        sa.Column(
+            "created_at",
+            sa.TIMESTAMP(timezone=True),
+            nullable=False,
+            server_default=sa.text("now()"),
+        ),
+        sa.Column(
+            "updated_at",
+            sa.TIMESTAMP(timezone=True),
+            nullable=False,
+            server_default=sa.text("now()"),
+        ),
+        sa.UniqueConstraint(
+            "connection_id", "parent_ref",
+            name="uq_cloud_folder_states_connection_parent",
+        ),
+    )
+
+    op.create_index("ix_cloud_folder_states_connection", "cloud_folder_states", ["connection_id"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_cloud_folder_states_connection", table_name="cloud_folder_states")
+    op.drop_table("cloud_folder_states")
+    op.drop_table("cloud_item_topics")
+    op.drop_index("ix_cloud_items_connection_parent", table_name="cloud_items")
+    op.drop_index("ix_cloud_items_connection_id", table_name="cloud_items")
+    op.drop_index("ix_cloud_items_user_id", table_name="cloud_items")
+    op.drop_table("cloud_items")
+    op.drop_column("cloud_connections", "display_name_override")

backend/services/cloud_items.py

@@ -0,0 +1,282 @@
+"""
+Owner-scoped cloud metadata reconciliation service — Phase 12.
+
+All functions operate within strict (user_id, connection_id) ownership boundaries.
+No function here raises FastAPI HTTPException — domain exceptions only.
+No function calls the quota service or alters quotas.used_bytes.
+
+Domain exceptions:
+  ConnectionNotFound — connection does not exist or belongs to a different user.
+  CloudItemNotFound  — cloud item does not exist for the given owner/connection.
+"""
+from __future__ import annotations
+
+import uuid
+from datetime import datetime, timezone
+from typing import Optional, Sequence
+
+from sqlalchemy import select, update
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from db.models import CloudConnection, CloudFolderState, CloudItem
+from storage.cloud_base import CloudListing, CloudResource
+
+
+# ── Domain exceptions ─────────────────────────────────────────────────────────
+
+class ConnectionNotFound(ValueError):
+    """Connection does not exist or belongs to a different user."""
+
+
+class CloudItemNotFound(ValueError):
+    """Cloud item does not exist for the given owner/connection."""
+
+
+# ── Connection resolution ─────────────────────────────────────────────────────
+
+async def resolve_owned_connection(
+    session: AsyncSession,
+    *,
+    connection_id: str,
+    user_id: str,
+) -> CloudConnection:
+    """Return the CloudConnection owned by user_id or raise ConnectionNotFound.
+
+    Never returns a connection belonging to another user.
+    """
+    result = await session.execute(
+        select(CloudConnection).where(
+            CloudConnection.id == connection_id,
+            CloudConnection.user_id == user_id,
+        )
+    )
+    conn = result.scalars().first()
+    if conn is None:
+        raise ConnectionNotFound(
+            f"Connection {connection_id!r} not found for user {user_id!r}"
+        )
+    return conn
+
+
+# ── Item listing ──────────────────────────────────────────────────────────────
+
+async def list_cloud_children(
+    session: AsyncSession,
+    *,
+    user_id: str,
+    connection_id: str,
+    parent_ref: Optional[str],
+) -> Sequence[CloudItem]:
+    """Return non-deleted cloud items matching (user_id, connection_id, parent_ref).
+
+    parent_ref=None matches items whose parent_ref is NULL (root children where
+    parent is not tracked as a ref string).
+    """
+    stmt = select(CloudItem).where(
+        CloudItem.user_id == user_id,
+        CloudItem.connection_id == connection_id,
+        CloudItem.deleted_at.is_(None),
+    )
+    if parent_ref is None:
+        stmt = stmt.where(CloudItem.parent_ref.is_(None))
+    else:
+        stmt = stmt.where(CloudItem.parent_ref == parent_ref)
+
+    result = await session.execute(stmt)
+    return result.scalars().all()
+
+
+# ── Item upsert ───────────────────────────────────────────────────────────────
+
+async def upsert_cloud_item(
+    session: AsyncSession,
+    *,
+    user_id: str,
+    resource: CloudResource,
+) -> CloudItem:
+    """Insert or update a CloudItem for the given normalized resource.
+
+    The DocuVault CloudItem UUID is preserved across provider rename/move:
+    if a row already exists for (connection_id, provider_item_id), its id
+    is retained and metadata fields are updated in place.
+
+    user_id is always set from the caller — never from the resource alone —
+    to enforce the owner boundary.
+    """
+    connection_id = str(resource.connection_id)
+
+    result = await session.execute(
+        select(CloudItem).where(
+            CloudItem.connection_id == connection_id,
+            CloudItem.provider_item_id == resource.provider_item_id,
+        )
+    )
+    existing = result.scalars().first()
+
+    now = datetime.now(timezone.utc)
+
+    if existing is not None:
+        # Update metadata, preserve UUID (stable identity across rename/move)
+        existing.name = resource.name
+        existing.kind = resource.kind
+        existing.parent_ref = resource.parent_ref
+        existing.content_type = resource.content_type
+        existing.provider_size = resource.size
+        existing.etag = resource.etag
+        existing.modified_at = resource.modified_at
+        existing.last_seen_at = now
+        existing.deleted_at = None  # un-delete if previously soft-deleted
+        existing.updated_at = now
+        await session.flush()
+        return existing
+    else:
+        item = CloudItem(
+            id=str(uuid.uuid4()),
+            user_id=user_id,
+            connection_id=connection_id,
+            provider_item_id=resource.provider_item_id,
+            parent_ref=resource.parent_ref,
+            name=resource.name,
+            kind=resource.kind,
+            content_type=resource.content_type,
+            provider_size=resource.size,
+            etag=resource.etag,
+            modified_at=resource.modified_at,
+            last_seen_at=now,
+            analysis_status="pending",
+            semantic_index_status="none",
+        )
+        session.add(item)
+        await session.flush()
+        return item
+
+
+# ── Listing reconciliation ────────────────────────────────────────────────────
+
+async def reconcile_cloud_listing(
+    session: AsyncSession,
+    *,
+    user_id: str,
+    connection_id: str,
+    parent_ref: Optional[str],
+    listing: CloudListing,
+) -> None:
+    """Reconcile a provider listing against durable cloud_items rows.
+
+    For each item in listing.items: upsert metadata.
+
+    Soft-deletion of missing children is ONLY performed when:
+      - listing.complete is True (authoritative, full listing)
+
+    Incomplete or failed listings (complete=False) must NEVER mark retained
+    rows deleted — provider bytes remain the source of truth.
+
+    This function never calls the quota service (D-18).
+    """
+    seen_provider_ids: set[str] = set()
+
+    for resource in listing.items:
+        await upsert_cloud_item(session, user_id=user_id, resource=resource)
+        seen_provider_ids.add(resource.provider_item_id)
+
+    if listing.complete:
+        # Soft-delete items not present in the complete listing
+        stmt = select(CloudItem).where(
+            CloudItem.user_id == user_id,
+            CloudItem.connection_id == connection_id,
+            CloudItem.deleted_at.is_(None),
+        )
+        if parent_ref is None:
+            stmt = stmt.where(CloudItem.parent_ref.is_(None))
+        else:
+            stmt = stmt.where(CloudItem.parent_ref == parent_ref)
+
+        result = await session.execute(stmt)
+        existing_items = result.scalars().all()
+
+        now = datetime.now(timezone.utc)
+        for item in existing_items:
+            if item.provider_item_id not in seen_provider_ids:
+                item.deleted_at = now
+                item.updated_at = now
+
+        await session.flush()
+
+
+# ── Folder state helpers ──────────────────────────────────────────────────────
+
+async def get_or_create_folder_state(
+    session: AsyncSession,
+    *,
+    user_id: str,
+    connection_id: str,
+    parent_ref: str,
+) -> CloudFolderState:
+    """Return the CloudFolderState for (connection_id, parent_ref), creating if absent.
+
+    Idempotent: repeated calls with the same arguments return the same row.
+    parent_ref='' represents the connection root.
+    """
+    result = await session.execute(
+        select(CloudFolderState).where(
+            CloudFolderState.connection_id == connection_id,
+            CloudFolderState.parent_ref == parent_ref,
+        )
+    )
+    existing = result.scalars().first()
+    if existing is not None:
+        return existing
+
+    fs = CloudFolderState(
+        id=str(uuid.uuid4()),
+        user_id=user_id,
+        connection_id=connection_id,
+        parent_ref=parent_ref,
+        refresh_state="fresh",
+    )
+    session.add(fs)
+    await session.flush()
+    return fs
+
+
+async def update_folder_state(
+    session: AsyncSession,
+    *,
+    user_id: str,
+    connection_id: str,
+    parent_ref: str,
+    refresh_state: str,
+    last_refreshed_at: Optional[datetime] = None,
+    error_code: Optional[str] = None,
+    error_message: Optional[str] = None,
+) -> CloudFolderState:
+    """Update the refresh state for a folder.
+
+    Transitions: refreshing → fresh (success) or warning (failure).
+    On success: set last_refreshed_at.
+    On failure: retain last_refreshed_at, set error_code/message with controlled values.
+
+    error_code and error_message must be controlled service values — never raw
+    provider exception text.
+    """
+    fs = await get_or_create_folder_state(
+        session, user_id=user_id, connection_id=connection_id, parent_ref=parent_ref
+    )
+    now = datetime.now(timezone.utc)
+    fs.refresh_state = refresh_state
+    fs.updated_at = now
+
+    if last_refreshed_at is not None:
+        fs.last_refreshed_at = last_refreshed_at
+    elif refresh_state == "fresh":
+        fs.last_refreshed_at = now
+
+    if refresh_state == "fresh":
+        fs.error_code = None
+        fs.error_message = None
+    elif error_code is not None:
+        fs.error_code = error_code
+        fs.error_message = error_message
+
+    await session.flush()
+    return fs

backend/tests/test_cloud_items.py

@@ -0,0 +1,528 @@
+"""
+Tests for cloud item metadata schema, model isolation, and reconciliation service.
+
+Task 2 tests (model/schema/isolation/quota):
+  - ORM model fields match migration schema expectations.
+  - Metadata, extracted text, topic links, and semantic data persist without
+    requiring an object key or byte cache row.
+  - Identical provider item IDs coexist across different connections.
+  - Foreign-owner queries do not return rows for other users.
+  - Provider size never alters Quota.used_bytes.
+  - Root folder state can be represented without a CloudItem parent row.
+
+Task 3 tests (reconciliation service):
+  - Repeated reconciliation creates no duplicate rows.
+  - Rename/move updates metadata without changing the CloudItem UUID.
+  - Incomplete or failed listing cannot mark unseen children deleted.
+  - All queries include both owner and connection scope.
+  - Idempotent repeated reconciliation.
+"""
+from __future__ import annotations
+
+import uuid
+from datetime import datetime, timezone
+from typing import Optional
+
+import pytest
+import pytest_asyncio
+from sqlalchemy import String, Text, event
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+from sqlalchemy.pool import StaticPool
+from sqlalchemy.dialects.postgresql import UUID, INET, JSONB
+
+from db.models import Base, CloudItem, CloudItemTopic, CloudFolderState, CloudConnection, Topic, User, Quota
+from storage.cloud_base import CloudListing, CloudResource, CloudCapability, STATE_SUPPORTED
+from services.cloud_items import (
+    resolve_owned_connection,
+    list_cloud_children,
+    upsert_cloud_item,
+    reconcile_cloud_listing,
+    get_or_create_folder_state,
+    update_folder_state,
+    CloudItemNotFound,
+    ConnectionNotFound,
+)
+
+
+# ── SQLite test engine setup ──────────────────────────────────────────────────
+
+@pytest_asyncio.fixture
+async def db_session():
+    """In-memory SQLite session with PostgreSQL-type shims."""
+    engine = create_async_engine(
+        "sqlite+aiosqlite:///:memory:",
+        connect_args={"check_same_thread": False},
+        poolclass=StaticPool,
+    )
+
+    # Shim PostgreSQL types to SQLite-compatible equivalents
+    from sqlalchemy import event as sa_event
+
+    @sa_event.listens_for(engine.sync_engine, "connect")
+    def set_sqlite_pragma(dbapi_connection, connection_record):
+        cursor = dbapi_connection.cursor()
+        cursor.execute("PRAGMA foreign_keys=ON")
+        cursor.close()
+
+    # Patch dialect-specific column types before table creation
+    import sqlalchemy.dialects.postgresql as pg
+    _orig_uuid_init = pg.UUID.__init__
+
+    def _patch_columns(metadata):
+        for table in metadata.tables.values():
+            for col in table.columns:
+                if isinstance(col.type, pg.UUID):
+                    col.type = String(36)
+                elif isinstance(col.type, pg.INET):
+                    col.type = String(45)
+                elif isinstance(col.type, pg.JSONB):
+                    col.type = Text()
+
+    async with engine.begin() as conn:
+        _patch_columns(Base.metadata)
+        await conn.run_sync(Base.metadata.create_all)
+
+    session_factory = async_sessionmaker(engine, expire_on_commit=False)
+    async with session_factory() as session:
+        yield session
+
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.drop_all)
+    await engine.dispose()
+
+
+# ── Helpers ───────────────────────────────────────────────────────────────────
+
+def _user_id() -> str:
+    return str(uuid.uuid4())
+
+
+def _conn_id() -> str:
+    return str(uuid.uuid4())
+
+
+def _item_id() -> str:
+    return str(uuid.uuid4())
+
+
+async def _make_user(session: AsyncSession, user_id: Optional[str] = None) -> str:
+    uid = user_id or _user_id()
+    u = User(
+        id=uid,
+        handle=f"user_{uid[:8]}",
+        email=f"{uid[:8]}@example.com",
+        password_hash="hash",
+        role="user",
+    )
+    session.add(u)
+    await session.flush()
+    return uid
+
+
+async def _make_connection(
+    session: AsyncSession, user_id: str, conn_id: Optional[str] = None
+) -> str:
+    cid = conn_id or _conn_id()
+    conn = CloudConnection(
+        id=cid,
+        user_id=user_id,
+        provider="onedrive",
+        display_name="My OneDrive",
+        credentials_enc="enc",
+        status="ACTIVE",
+    )
+    session.add(conn)
+    await session.flush()
+    return cid
+
+
+def _cloud_resource(
+    connection_id: str,
+    user_id: str,
+    provider_item_id: str = "item-001",
+    name: str = "test.pdf",
+    kind: str = "file",
+    parent_ref: Optional[str] = None,
+    size: Optional[int] = None,
+    etag: Optional[str] = None,
+) -> CloudResource:
+    return CloudResource(
+        id=uuid.uuid4(),
+        provider_item_id=provider_item_id,
+        connection_id=uuid.UUID(connection_id),
+        user_id=uuid.UUID(user_id),
+        name=name,
+        kind=kind,
+        parent_ref=parent_ref,
+        size=size,
+        etag=etag,
+    )
+
+
+# ── Task 2: model/schema tests ────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_model_cloud_item_basic_fields(db_session: AsyncSession):
+    """CloudItem persists metadata without requiring an object key."""
+    uid = await _make_user(db_session)
+    cid = await _make_connection(db_session, uid)
+    item_id = _item_id()
+
+    item = CloudItem(
+        id=item_id,
+        user_id=uid,
+        connection_id=cid,
+        provider_item_id="driveitem-123",
+        name="report.pdf",
+        kind="file",
+        content_type="application/pdf",
+        provider_size=204800,
+        etag="etag-v1",
+    )
+    db_session.add(item)
+    await db_session.flush()
+    await db_session.refresh(item)
+
+    assert item.id == item_id
+    assert item.provider_item_id == "driveitem-123"
+    assert item.name == "report.pdf"
+    assert item.kind == "file"
+    assert item.provider_size == 204800
+    assert item.analysis_status == "pending"
+    assert item.semantic_index_status == "none"
+    assert item.deleted_at is None
+
+
+@pytest.mark.asyncio
+async def test_model_cloud_item_extracted_text_without_object_key(db_session: AsyncSession):
+    """Extracted text and semantic data persist with no MinIO object key field."""
+    uid = await _make_user(db_session)
+    cid = await _make_connection(db_session, uid)
+
+    item = CloudItem(
+        id=_item_id(),
+        user_id=uid,
+        connection_id=cid,
+        provider_item_id="item-text",
+        name="notes.txt",
+        kind="file",
+        extracted_text="Sample extracted text",
+        semantic_index_status="indexed",
+        # No object_key field exists — provider bytes are not retained here
+    )
+    db_session.add(item)
+    await db_session.flush()
+    await db_session.refresh(item)
+
+    assert item.extracted_text == "Sample extracted text"
+    assert item.semantic_index_status == "indexed"
+    # Confirm no object_key attribute at all
+    assert not hasattr(item, "object_key")
+
+
+@pytest.mark.asyncio
+async def test_model_cloud_item_topics_no_document_row(db_session: AsyncSession):
+    """CloudItemTopic links item to topic without requiring a Document row."""
+    uid = await _make_user(db_session)
+    cid = await _make_connection(db_session, uid)
+
+    topic = Topic(id=_item_id(), user_id=uid, name="Finance", description="", color="#ff0000")
+    db_session.add(topic)
+    await db_session.flush()
+
+    item = CloudItem(
+        id=_item_id(),
+        user_id=uid,
+        connection_id=cid,
+        provider_item_id="item-topic",
+        name="budget.xlsx",
+        kind="file",
+    )
+    db_session.add(item)
+    await db_session.flush()
+
+    link = CloudItemTopic(cloud_item_id=item.id, topic_id=topic.id)
+    db_session.add(link)
+    await db_session.flush()
+
+    # Verify link persisted
+    from sqlalchemy import select
+    result = await db_session.execute(
+        select(CloudItemTopic).where(CloudItemTopic.cloud_item_id == item.id)
+    )
+    rows = result.scalars().all()
+    assert len(rows) == 1
+    assert rows[0].topic_id == topic.id
+
+
+@pytest.mark.asyncio
+async def test_model_schema_isolation_same_provider_item_different_connections(db_session: AsyncSession):
+    """Identical provider_item_id can coexist across different connections."""
+    uid = await _make_user(db_session)
+    cid1 = await _make_connection(db_session, uid)
+    cid2 = await _make_connection(db_session, uid)
+    shared_pid = "shared-item-001"
+
+    item1 = CloudItem(
+        id=_item_id(), user_id=uid, connection_id=cid1,
+        provider_item_id=shared_pid, name="file.pdf", kind="file",
+    )
+    item2 = CloudItem(
+        id=_item_id(), user_id=uid, connection_id=cid2,
+        provider_item_id=shared_pid, name="file.pdf", kind="file",
+    )
+    db_session.add_all([item1, item2])
+    await db_session.flush()
+
+    from sqlalchemy import select
+    result = await db_session.execute(
+        select(CloudItem).where(CloudItem.provider_item_id == shared_pid)
+    )
+    rows = result.scalars().all()
+    assert len(rows) == 2
+
+
+@pytest.mark.asyncio
+async def test_model_schema_foreign_owner_query_returns_nothing(db_session: AsyncSession):
+    """Items for user A are not returned when querying for user B."""
+    uid_a = await _make_user(db_session)
+    uid_b = await _make_user(db_session)
+    cid_a = await _make_connection(db_session, uid_a)
+
+    item = CloudItem(
+        id=_item_id(), user_id=uid_a, connection_id=cid_a,
+        provider_item_id="item-a", name="private.pdf", kind="file",
+    )
+    db_session.add(item)
+    await db_session.flush()
+
+    from sqlalchemy import select
+    result = await db_session.execute(
+        select(CloudItem).where(
+            CloudItem.user_id == uid_b,
+            CloudItem.provider_item_id == "item-a",
+        )
+    )
+    assert result.scalars().first() is None
+
+
+@pytest.mark.asyncio
+async def test_model_quota_unchanged_after_item_upsert(db_session: AsyncSession):
+    """Upserting a cloud item with provider_size must not alter Quota.used_bytes."""
+    uid = await _make_user(db_session)
+    cid = await _make_connection(db_session, uid)
+
+    quota = Quota(user_id=uid, limit_bytes=104857600, used_bytes=0)
+    db_session.add(quota)
+    await db_session.flush()
+
+    item = CloudItem(
+        id=_item_id(), user_id=uid, connection_id=cid,
+        provider_item_id="large-file", name="video.mp4", kind="file",
+        provider_size=5_000_000_000,  # 5 GB provider-reported size
+    )
+    db_session.add(item)
+    await db_session.flush()
+    await db_session.refresh(quota)
+
+    # Quota must remain unchanged
+    assert quota.used_bytes == 0
+
+
+@pytest.mark.asyncio
+async def test_model_folder_state_root_no_parent_item(db_session: AsyncSession):
+    """Root folder state can be created with parent_ref='' without a CloudItem row."""
+    uid = await _make_user(db_session)
+    cid = await _make_connection(db_session, uid)
+
+    fs = CloudFolderState(
+        id=_item_id(),
+        user_id=uid,
+        connection_id=cid,
+        parent_ref="",  # root
+        refresh_state="fresh",
+    )
+    db_session.add(fs)
+    await db_session.flush()
+    await db_session.refresh(fs)
+
+    assert fs.parent_ref == ""
+    assert fs.refresh_state == "fresh"
+    assert fs.last_refreshed_at is None
+
+
+# ── Task 3: reconciliation service tests ─────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_service_resolve_owned_connection_found(db_session: AsyncSession):
+    uid = await _make_user(db_session)
+    cid = await _make_connection(db_session, uid)
+
+    conn = await resolve_owned_connection(db_session, connection_id=cid, user_id=uid)
+    assert str(conn.id) == cid
+
+
+@pytest.mark.asyncio
+async def test_service_resolve_owned_connection_wrong_owner_raises(db_session: AsyncSession):
+    uid_a = await _make_user(db_session)
+    uid_b = await _make_user(db_session)
+    cid = await _make_connection(db_session, uid_a)
+
+    with pytest.raises(ConnectionNotFound):
+        await resolve_owned_connection(db_session, connection_id=cid, user_id=uid_b)
+
+
+@pytest.mark.asyncio
+async def test_service_upsert_creates_item(db_session: AsyncSession):
+    uid = await _make_user(db_session)
+    cid = await _make_connection(db_session, uid)
+    resource = _cloud_resource(cid, uid, provider_item_id="new-001", name="doc.pdf")
+
+    item = await upsert_cloud_item(db_session, user_id=uid, resource=resource)
+    assert item.provider_item_id == "new-001"
+    assert item.name == "doc.pdf"
+
+
+@pytest.mark.asyncio
+async def test_service_upsert_idempotent_same_uuid(db_session: AsyncSession):
+    """Repeated upsert with unchanged data creates no duplicate row."""
+    uid = await _make_user(db_session)
+    cid = await _make_connection(db_session, uid)
+    resource = _cloud_resource(cid, uid, provider_item_id="idem-001")
+
+    item1 = await upsert_cloud_item(db_session, user_id=uid, resource=resource)
+    item2 = await upsert_cloud_item(db_session, user_id=uid, resource=resource)
+
+    assert item1.id == item2.id  # same DocuVault UUID preserved
+
+
+@pytest.mark.asyncio
+async def test_service_upsert_rename_preserves_uuid(db_session: AsyncSession):
+    """Rename updates name but keeps the same CloudItem UUID."""
+    uid = await _make_user(db_session)
+    cid = await _make_connection(db_session, uid)
+    resource = _cloud_resource(cid, uid, provider_item_id="rename-001", name="old.pdf")
+
+    item1 = await upsert_cloud_item(db_session, user_id=uid, resource=resource)
+    original_uuid = item1.id
+
+    renamed = _cloud_resource(cid, uid, provider_item_id="rename-001", name="new.pdf")
+    item2 = await upsert_cloud_item(db_session, user_id=uid, resource=renamed)
+
+    assert item2.id == original_uuid
+    assert item2.name == "new.pdf"
+
+
+@pytest.mark.asyncio
+async def test_service_list_children_owner_scoped(db_session: AsyncSession):
+    """list_cloud_children only returns items for the correct (user, connection, parent)."""
+    uid_a = await _make_user(db_session)
+    uid_b = await _make_user(db_session)
+    cid_a = await _make_connection(db_session, uid_a)
+
+    res = _cloud_resource(cid_a, uid_a, provider_item_id="child-001", parent_ref="parent-ref")
+    await upsert_cloud_item(db_session, user_id=uid_a, resource=res)
+
+    # User A sees the item
+    children_a = await list_cloud_children(
+        db_session, user_id=uid_a, connection_id=cid_a, parent_ref="parent-ref"
+    )
+    assert len(children_a) == 1
+
+    # User B does not
+    children_b = await list_cloud_children(
+        db_session, user_id=uid_b, connection_id=cid_a, parent_ref="parent-ref"
+    )
+    assert len(children_b) == 0
+
+
+@pytest.mark.asyncio
+async def test_service_reconcile_complete_marks_missing_deleted(db_session: AsyncSession):
+    """Complete listing marks items not in the listing as deleted."""
+    uid = await _make_user(db_session)
+    cid = await _make_connection(db_session, uid)
+
+    # Pre-existing item
+    existing = _cloud_resource(cid, uid, provider_item_id="keep-001", parent_ref="root")
+    gone = _cloud_resource(cid, uid, provider_item_id="gone-001", parent_ref="root")
+    await upsert_cloud_item(db_session, user_id=uid, resource=existing)
+    await upsert_cloud_item(db_session, user_id=uid, resource=gone)
+
+    # Reconcile with only "keep-001" in the complete listing
+    listing = CloudListing(items=[existing], complete=True)
+    await reconcile_cloud_listing(
+        db_session, user_id=uid, connection_id=cid, parent_ref="root", listing=listing
+    )
+
+    from sqlalchemy import select
+    result = await db_session.execute(
+        select(CloudItem).where(CloudItem.provider_item_id == "gone-001")
+    )
+    gone_item = result.scalars().first()
+    assert gone_item is not None
+    assert gone_item.deleted_at is not None
+
+
+@pytest.mark.asyncio
+async def test_service_reconcile_incomplete_does_not_delete(db_session: AsyncSession):
+    """Incomplete (partial/failed) listing must not mark unseen children deleted."""
+    uid = await _make_user(db_session)
+    cid = await _make_connection(db_session, uid)
+
+    item = _cloud_resource(cid, uid, provider_item_id="retain-001", parent_ref="root")
+    await upsert_cloud_item(db_session, user_id=uid, resource=item)
+
+    # Reconcile with empty incomplete listing
+    listing = CloudListing(items=[], complete=False)
+    await reconcile_cloud_listing(
+        db_session, user_id=uid, connection_id=cid, parent_ref="root", listing=listing
+    )
+
+    from sqlalchemy import select
+    result = await db_session.execute(
+        select(CloudItem).where(CloudItem.provider_item_id == "retain-001")
+    )
+    retained = result.scalars().first()
+    assert retained is not None
+    assert retained.deleted_at is None
+
+
+@pytest.mark.asyncio
+async def test_service_folder_state_create_and_update(db_session: AsyncSession):
+    uid = await _make_user(db_session)
+    cid = await _make_connection(db_session, uid)
+
+    fs = await get_or_create_folder_state(db_session, user_id=uid, connection_id=cid, parent_ref="")
+    assert fs.refresh_state == "fresh"
+
+    await update_folder_state(
+        db_session,
+        user_id=uid,
+        connection_id=cid,
+        parent_ref="",
+        refresh_state="warning",
+        error_code="token_expired",
+        error_message="Re-authenticate to refresh.",
+    )
+
+    from sqlalchemy import select
+    result = await db_session.execute(
+        select(CloudFolderState).where(
+            CloudFolderState.connection_id == cid,
+            CloudFolderState.parent_ref == "",
+        )
+    )
+    updated = result.scalars().first()
+    assert updated.refresh_state == "warning"
+    assert updated.error_code == "token_expired"
+
+
+@pytest.mark.asyncio
+async def test_service_folder_state_idempotent(db_session: AsyncSession):
+    """get_or_create is idempotent — no duplicate rows."""
+    uid = await _make_user(db_session)
+    cid = await _make_connection(db_session, uid)
+
+    fs1 = await get_or_create_folder_state(db_session, user_id=uid, connection_id=cid, parent_ref="")
+    fs2 = await get_or_create_folder_state(db_session, user_id=uid, connection_id=cid, parent_ref="")
+    assert fs1.id == fs2.id