feat(14-03): add cache orchestration functions and tests

- retain_or_reuse_cache_entry: reuse non-evicted entry or reactivate evicted row
- pin_cache_entry / release_cache_entry: pin_count lifecycle management
- update_cache_access: touch last_accessed_at for LRU ordering
- Ownership assertions in all new service functions (T-14-06)
- 9 new integration tests covering retain/reuse, pin/release, and isolation

backend/services/cloud_cache.py

@@ -21,7 +21,8 @@ References:
 
 ── Phase 14: durable byte cache (cloud_byte_cache_entries) ─────────────────────
 Exports create_cache_entry, evict_lru_entries, list_cache_entries,
-increment_quota_for_cache, decrement_quota_for_cache for use by Celery tasks
+increment_quota_for_cache, decrement_quota_for_cache, retain_or_reuse_cache_entry,
+pin_cache_entry, release_cache_entry, update_cache_access for use by Celery tasks
 and the analysis API.
 
 Also re-exports compute_version_key from services.cloud_analysis_versioning so
@@ -323,6 +324,232 @@ async def decrement_quota_for_cache(
     )
 
 
+async def retain_or_reuse_cache_entry(
+    session: AsyncSession,
+    *,
+    user_id: uuid.UUID,
+    connection_id: uuid.UUID,
+    cloud_item_id: uuid.UUID,
+    provider_item_id: str,
+    version_key: str,
+    object_key: str,
+    content_type: Optional[str] = None,
+    size_bytes: int,
+    content_hash: Optional[str] = None,
+) -> tuple[CloudByteCacheEntry, bool]:
+    """Look up a non-evicted cache entry matching the version_key; create one if absent.
+
+    Implements the Phase 14 cache lifecycle step 3 (PATTERNS.md): reuse matching
+    non-evicted entry before hydrating from provider.
+
+    Ownership is strictly enforced — queries filter by (user_id, connection_id,
+    cloud_item_id, version_key) so a foreign user can never reuse another user's entry.
+
+    When creating a new entry this function does NOT call increment_quota_for_cache.
+    The caller is responsible for reserving quota atomically before retaining bytes
+    and calling increment_quota_for_cache before this function.
+
+    Args:
+        session:         Active async SQLAlchemy session.
+        user_id:         Owner UUID.
+        connection_id:   Cloud connection UUID.
+        cloud_item_id:   CloudItem row UUID.
+        provider_item_id: Provider item id snapshot for audit.
+        version_key:     Idempotency key from compute_version_key.
+        object_key:      Private MinIO object key — UUID-based, never exposed.
+        content_type:    Optional MIME type.
+        size_bytes:      Byte count stored in MinIO.
+        content_hash:    Optional post-hydration content hash.
+
+    Returns:
+        Tuple (entry, created_new) where created_new=True if a new row was inserted,
+        False if an existing non-evicted entry was reused.
+    """
+    uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id))
+    cid = connection_id if isinstance(connection_id, uuid.UUID) else uuid.UUID(str(connection_id))
+    iid = cloud_item_id if isinstance(cloud_item_id, uuid.UUID) else uuid.UUID(str(cloud_item_id))
+
+    now = datetime.now(timezone.utc)
+
+    # Step 3: check for existing entry (evicted or not) for this version key.
+    # The unique constraint on (user_id, connection_id, cloud_item_id, version_key) means
+    # at most one row exists per version key.  We handle both states:
+    #   - non-evicted → reuse (touch last_accessed_at)
+    #   - evicted → reactivate (clear evicted_at, update object_key + size)
+    stmt = (
+        select(CloudByteCacheEntry)
+        .where(
+            CloudByteCacheEntry.user_id == uid,
+            CloudByteCacheEntry.connection_id == cid,
+            CloudByteCacheEntry.cloud_item_id == iid,
+            CloudByteCacheEntry.version_key == version_key,
+        )
+        .limit(1)
+    )
+    result = await session.execute(stmt)
+    existing = result.scalars().first()
+
+    if existing is not None:
+        if existing.evicted_at is None:
+            # Active entry — reuse it
+            existing.last_accessed_at = now
+            existing.updated_at = now
+            await session.flush()
+            return existing, False
+        else:
+            # Evicted entry — reactivate with fresh bytes and updated object key
+            existing.evicted_at = None
+            existing.object_key = object_key
+            existing.size_bytes = size_bytes
+            existing.content_type = content_type
+            existing.content_hash = content_hash
+            existing.last_accessed_at = now
+            existing.updated_at = now
+            await session.flush()
+            return existing, True
+
+    # No existing entry — create a new one
+    # Quota must be reserved by caller before bytes are stored in MinIO
+    new_entry = await create_cache_entry(
+        session,
+        user_id=uid,
+        connection_id=cid,
+        cloud_item_id=iid,
+        provider_item_id=provider_item_id,
+        version_key=version_key,
+        object_key=object_key,
+        content_type=content_type,
+        size_bytes=size_bytes,
+        content_hash=content_hash,
+    )
+    return new_entry, True
+
+
+async def pin_cache_entry(
+    session: AsyncSession,
+    *,
+    entry_id: uuid.UUID,
+    user_id: uuid.UUID,
+) -> CloudByteCacheEntry:
+    """Increment pin_count to mark an active preview or download lease.
+
+    Pinned entries are protected from LRU eviction (PATTERNS.md step 7).
+    Ownership is asserted — user_id must match the entry's user_id.
+
+    Args:
+        session:  Active async SQLAlchemy session.
+        entry_id: CloudByteCacheEntry UUID.
+        user_id:  Authenticated user UUID — must own the entry.
+
+    Returns:
+        Updated CloudByteCacheEntry.
+
+    Raises:
+        ValueError: Entry not found or does not belong to user_id.
+    """
+    uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id))
+    eid = entry_id if isinstance(entry_id, uuid.UUID) else uuid.UUID(str(entry_id))
+
+    result = await session.execute(
+        select(CloudByteCacheEntry).where(
+            CloudByteCacheEntry.id == eid,
+            CloudByteCacheEntry.user_id == uid,
+        )
+    )
+    entry = result.scalars().first()
+    if entry is None:
+        raise ValueError(f"Cache entry {eid} not found for user {uid}")
+
+    now = datetime.now(timezone.utc)
+    entry.pin_count += 1
+    entry.last_accessed_at = now
+    entry.updated_at = now
+    await session.flush()
+    return entry
+
+
+async def release_cache_entry(
+    session: AsyncSession,
+    *,
+    entry_id: uuid.UUID,
+    user_id: uuid.UUID,
+) -> CloudByteCacheEntry:
+    """Decrement pin_count when a preview or download lease ends.
+
+    Clamped to 0 — defensive against double-release (PATTERNS.md step 8).
+
+    Args:
+        session:  Active async SQLAlchemy session.
+        entry_id: CloudByteCacheEntry UUID.
+        user_id:  Authenticated user UUID — must own the entry.
+
+    Returns:
+        Updated CloudByteCacheEntry.
+
+    Raises:
+        ValueError: Entry not found or does not belong to user_id.
+    """
+    uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id))
+    eid = entry_id if isinstance(entry_id, uuid.UUID) else uuid.UUID(str(entry_id))
+
+    result = await session.execute(
+        select(CloudByteCacheEntry).where(
+            CloudByteCacheEntry.id == eid,
+            CloudByteCacheEntry.user_id == uid,
+        )
+    )
+    entry = result.scalars().first()
+    if entry is None:
+        raise ValueError(f"Cache entry {eid} not found for user {uid}")
+
+    now = datetime.now(timezone.utc)
+    entry.pin_count = max(0, entry.pin_count - 1)
+    entry.updated_at = now
+    await session.flush()
+    return entry
+
+
+async def update_cache_access(
+    session: AsyncSession,
+    *,
+    entry_id: uuid.UUID,
+    user_id: uuid.UUID,
+) -> CloudByteCacheEntry:
+    """Touch last_accessed_at to update LRU position.
+
+    Called after a successful cache hit to maintain accurate LRU ordering.
+
+    Args:
+        session:  Active async SQLAlchemy session.
+        entry_id: CloudByteCacheEntry UUID.
+        user_id:  Authenticated user UUID — must own the entry.
+
+    Returns:
+        Updated CloudByteCacheEntry.
+
+    Raises:
+        ValueError: Entry not found or does not belong to user_id.
+    """
+    uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id))
+    eid = entry_id if isinstance(entry_id, uuid.UUID) else uuid.UUID(str(entry_id))
+
+    result = await session.execute(
+        select(CloudByteCacheEntry).where(
+            CloudByteCacheEntry.id == eid,
+            CloudByteCacheEntry.user_id == uid,
+        )
+    )
+    entry = result.scalars().first()
+    if entry is None:
+        raise ValueError(f"Cache entry {eid} not found for user {uid}")
+
+    now = datetime.now(timezone.utc)
+    entry.last_accessed_at = now
+    entry.updated_at = now
+    await session.flush()
+    return entry
+
+
 # ── Phase 12: in-memory folder listing cache ──────────────────────────────────
 
 

backend/tests/test_cloud_cache.py

@@ -493,3 +493,287 @@ async def test_content_hash_not_required_during_version_key_computation():
 
     assert key is not None
     assert isinstance(key, str)
+
+
+# ─── retain_or_reuse_cache_entry ─────────────────────────────────────────────
+
+async def test_retain_or_reuse_returns_existing_entry(db_session):
+    """CACHE-03: retain_or_reuse_cache_entry returns an existing non-evicted entry."""
+    from services.cloud_cache import create_cache_entry, retain_or_reuse_cache_entry
+
+    user_id = uuid.uuid4()
+    conn_id = uuid.uuid4()
+    item_id = uuid.uuid4()
+    version_key = "etag-shared-v1"
+
+    # Create the first entry as the existing cached bytes
+    first = await create_cache_entry(
+        session=db_session,
+        user_id=user_id,
+        connection_id=conn_id,
+        cloud_item_id=item_id,
+        provider_item_id="pitem-retain",
+        version_key=version_key,
+        object_key=f"cache/{user_id}/{uuid.uuid4()}.pdf",
+        content_type="application/pdf",
+        size_bytes=1024,
+    )
+
+    # retain_or_reuse must return the existing entry, not create a new one
+    reused, created_new = await retain_or_reuse_cache_entry(
+        session=db_session,
+        user_id=user_id,
+        connection_id=conn_id,
+        cloud_item_id=item_id,
+        provider_item_id="pitem-retain",
+        version_key=version_key,
+        object_key=f"cache/{user_id}/{uuid.uuid4()}-new.pdf",  # different key — should not be used
+        size_bytes=1024,
+    )
+
+    assert not created_new, "Should reuse existing entry, not create a new one"
+    assert str(reused.id) == str(first.id), "Must return the same DB row"
+
+
+async def test_retain_or_reuse_creates_new_entry_when_absent(db_session):
+    """CACHE-03: retain_or_reuse_cache_entry creates a new entry if none exists."""
+    from services.cloud_cache import retain_or_reuse_cache_entry
+
+    user_id = uuid.uuid4()
+    conn_id = uuid.uuid4()
+    item_id = uuid.uuid4()
+    new_key = f"cache/{user_id}/{uuid.uuid4()}.pdf"
+
+    entry, created_new = await retain_or_reuse_cache_entry(
+        session=db_session,
+        user_id=user_id,
+        connection_id=conn_id,
+        cloud_item_id=item_id,
+        provider_item_id="pitem-new",
+        version_key="etag-brand-new",
+        object_key=new_key,
+        size_bytes=2048,
+    )
+
+    assert created_new, "No existing entry — must create a new one"
+    assert entry.size_bytes == 2048
+    assert entry.user_id == user_id
+
+
+async def test_retain_or_reuse_reactivates_evicted_entry(db_session):
+    """CACHE-03: An evicted entry for the same version key is reactivated, not duplicated.
+
+    The unique constraint on (user_id, connection_id, cloud_item_id, version_key) means
+    inserting a new row is impossible.  retain_or_reuse_cache_entry reactivates the evicted
+    row (clears evicted_at) and updates the object_key to point to fresh bytes.
+    """
+    from services.cloud_cache import create_cache_entry, retain_or_reuse_cache_entry
+    from datetime import datetime, timezone
+
+    user_id = uuid.uuid4()
+    conn_id = uuid.uuid4()
+    item_id = uuid.uuid4()
+    version_key = "etag-evicted"
+
+    # Create and evict an entry
+    evicted = await create_cache_entry(
+        session=db_session,
+        user_id=user_id,
+        connection_id=conn_id,
+        cloud_item_id=item_id,
+        provider_item_id="pitem-evicted",
+        version_key=version_key,
+        object_key=f"cache/{user_id}/{uuid.uuid4()}.pdf",
+        size_bytes=512,
+    )
+    evicted.evicted_at = datetime.now(timezone.utc)
+    await db_session.flush()
+
+    fresh_key = f"cache/{user_id}/{uuid.uuid4()}-fresh.pdf"
+    reactivated, created_new = await retain_or_reuse_cache_entry(
+        session=db_session,
+        user_id=user_id,
+        connection_id=conn_id,
+        cloud_item_id=item_id,
+        provider_item_id="pitem-evicted",
+        version_key=version_key,
+        object_key=fresh_key,
+        size_bytes=512,
+    )
+
+    # created_new=True because we needed to rehydrate bytes (evicted entry was unavailable)
+    assert created_new, "Reactivated evicted entry must report created_new=True"
+    # Same DB row is reused (no duplicate due to unique constraint)
+    assert str(reactivated.id) == str(evicted.id)
+    # evicted_at must be cleared
+    assert reactivated.evicted_at is None, "Reactivated entry must have evicted_at=None"
+    # Object key updated to fresh bytes
+    assert reactivated.object_key == fresh_key
+
+
+async def test_retain_or_reuse_isolates_foreign_user(db_session):
+    """T-14-06: retain_or_reuse_cache_entry never returns another user's entry."""
+    from services.cloud_cache import create_cache_entry, retain_or_reuse_cache_entry
+
+    owner_id = uuid.uuid4()
+    foreign_id = uuid.uuid4()
+    conn_id = uuid.uuid4()
+    item_id = uuid.uuid4()
+    version_key = "etag-shared"
+
+    # Create entry for owner
+    owner_entry = await create_cache_entry(
+        session=db_session,
+        user_id=owner_id,
+        connection_id=conn_id,
+        cloud_item_id=item_id,
+        provider_item_id="pitem-owner",
+        version_key=version_key,
+        object_key=f"cache/{owner_id}/{uuid.uuid4()}.pdf",
+        size_bytes=4096,
+    )
+
+    # Foreign user must get a new entry, not the owner's
+    foreign_entry, created_new = await retain_or_reuse_cache_entry(
+        session=db_session,
+        user_id=foreign_id,
+        connection_id=conn_id,
+        cloud_item_id=item_id,
+        provider_item_id="pitem-owner",
+        version_key=version_key,
+        object_key=f"cache/{foreign_id}/{uuid.uuid4()}.pdf",
+        size_bytes=4096,
+    )
+
+    assert created_new, "Foreign user must not reuse another user's cache entry (T-14-06)"
+    assert str(foreign_entry.id) != str(owner_entry.id)
+
+
+# ─── pin_cache_entry / release_cache_entry ────────────────────────────────────
+
+async def test_pin_increments_pin_count(db_session):
+    """CACHE-04: pin_cache_entry increments pin_count to protect from eviction."""
+    from services.cloud_cache import create_cache_entry, pin_cache_entry
+
+    user_id = uuid.uuid4()
+    conn_id = uuid.uuid4()
+    item_id = uuid.uuid4()
+
+    entry = await create_cache_entry(
+        session=db_session,
+        user_id=user_id,
+        connection_id=conn_id,
+        cloud_item_id=item_id,
+        provider_item_id="pitem-pin",
+        version_key="etag-pin-v1",
+        object_key=f"cache/{user_id}/{uuid.uuid4()}.pdf",
+        size_bytes=1024,
+        pin_count=0,
+    )
+    assert entry.pin_count == 0
+
+    pinned = await pin_cache_entry(session=db_session, entry_id=entry.id, user_id=user_id)
+    assert pinned.pin_count == 1, "pin_cache_entry must increment pin_count"
+
+
+async def test_release_decrements_pin_count(db_session):
+    """CACHE-04: release_cache_entry decrements pin_count back to 0."""
+    from services.cloud_cache import create_cache_entry, pin_cache_entry, release_cache_entry
+
+    user_id = uuid.uuid4()
+    conn_id = uuid.uuid4()
+    item_id = uuid.uuid4()
+
+    entry = await create_cache_entry(
+        session=db_session,
+        user_id=user_id,
+        connection_id=conn_id,
+        cloud_item_id=item_id,
+        provider_item_id="pitem-release",
+        version_key="etag-release-v1",
+        object_key=f"cache/{user_id}/{uuid.uuid4()}.pdf",
+        size_bytes=1024,
+    )
+
+    await pin_cache_entry(session=db_session, entry_id=entry.id, user_id=user_id)
+    released = await release_cache_entry(session=db_session, entry_id=entry.id, user_id=user_id)
+    assert released.pin_count == 0, "release_cache_entry must decrement pin_count"
+
+
+async def test_release_clamps_to_zero(db_session):
+    """CACHE-04: release_cache_entry on an already-zero pin_count stays at 0 (defensive)."""
+    from services.cloud_cache import create_cache_entry, release_cache_entry
+
+    user_id = uuid.uuid4()
+    conn_id = uuid.uuid4()
+    item_id = uuid.uuid4()
+
+    entry = await create_cache_entry(
+        session=db_session,
+        user_id=user_id,
+        connection_id=conn_id,
+        cloud_item_id=item_id,
+        provider_item_id="pitem-clamp",
+        version_key="etag-clamp",
+        object_key=f"cache/{user_id}/{uuid.uuid4()}.pdf",
+        size_bytes=512,
+        pin_count=0,
+    )
+
+    released = await release_cache_entry(session=db_session, entry_id=entry.id, user_id=user_id)
+    assert released.pin_count == 0, "Double-release must not produce negative pin_count"
+
+
+async def test_pin_raises_on_foreign_entry(db_session):
+    """T-14-06: pin_cache_entry raises ValueError on foreign-user entry."""
+    from services.cloud_cache import create_cache_entry, pin_cache_entry
+
+    owner_id = uuid.uuid4()
+    foreign_id = uuid.uuid4()
+    conn_id = uuid.uuid4()
+
+    entry = await create_cache_entry(
+        session=db_session,
+        user_id=owner_id,
+        connection_id=conn_id,
+        cloud_item_id=uuid.uuid4(),
+        provider_item_id="pitem-foreign-pin",
+        version_key="etag-foreign",
+        object_key=f"cache/{owner_id}/{uuid.uuid4()}.pdf",
+        size_bytes=512,
+    )
+
+    with pytest.raises(ValueError, match="not found"):
+        await pin_cache_entry(session=db_session, entry_id=entry.id, user_id=foreign_id)
+
+
+# ─── update_cache_access ─────────────────────────────────────────────────────
+
+async def test_update_cache_access_touches_last_accessed(db_session):
+    """CACHE-04: update_cache_access refreshes last_accessed_at for LRU ordering."""
+    from services.cloud_cache import create_cache_entry, update_cache_access
+    from datetime import datetime, timezone, timedelta
+
+    user_id = uuid.uuid4()
+    conn_id = uuid.uuid4()
+
+    old_ts = datetime.now(timezone.utc) - timedelta(hours=3)
+    entry = await create_cache_entry(
+        session=db_session,
+        user_id=user_id,
+        connection_id=conn_id,
+        cloud_item_id=uuid.uuid4(),
+        provider_item_id="pitem-access",
+        version_key="etag-access",
+        object_key=f"cache/{user_id}/{uuid.uuid4()}.pdf",
+        size_bytes=256,
+        last_accessed_at=old_ts,
+    )
+
+    updated = await update_cache_access(
+        session=db_session, entry_id=entry.id, user_id=user_id
+    )
+
+    assert updated.last_accessed_at > old_ts, (
+        "update_cache_access must refresh last_accessed_at beyond the old timestamp"
+    )