---
phase: 03-document-migration-multi-user-isolation
plan: 02
type: execute
wave: 2
depends_on:
  - 03-01
files_modified:
  - backend/storage/base.py
  - backend/storage/minio_backend.py
  - backend/storage/__init__.py
  - backend/config.py
  - backend/api/documents.py
  - backend/api/auth.py
  - backend/services/storage.py
  - backend/tasks/document_tasks.py
  - backend/celery_app.py
  - docker-compose.yml
autonomous: true
requirements:
  - STORE-03
  - STORE-04
  - STORE-05
  - STORE-06
  - SEC-04

must_haves:
  truths:
    - "Frontend can request a presigned PUT URL targeting a browser-resolvable hostname (not 'minio:9000') for a newly created pending Document row"
    - "After the browser PUT, /confirm reads the authoritative file size from MinIO stat_object and atomically updates the user's quota, returning 413 with {used_bytes, limit_bytes, rejected_bytes} on overflow"
    - "Two concurrent /confirm calls that would together exceed the quota result in exactly one 200 and one 413"
    - "Document delete decrements quota using GREATEST(0, used_bytes - delta) atomically"
    - "Authenticated user can read their own quota via GET /api/auth/me/quota"
    - "Celery beat runs cleanup_abandoned_uploads every 30 minutes, deleting pending Document rows older than 1 hour and removing their MinIO objects"
    - "Browser PUT requests succeed against MinIO because MINIO_API_CORS_ALLOW_ORIGIN is set to the configured CORS origin"
  artifacts:
    - path: "backend/storage/base.py"
      provides: "StorageBackend ABC with generate_presigned_put_url and stat_object abstract methods"
      contains: "generate_presigned_put_url"
    - path: "backend/storage/minio_backend.py"
      provides: "MinIOBackend dual-client (internal + public) + generate_presigned_put_url + stat_object methods"
      contains: "self._public_client"
    - path: "backend/storage/__init__.py"
      provides: "get_storage_backend factory passes public_endpoint to MinIOBackend"
      contains: "public_endpoint"
    - path: "backend/api/documents.py"
      provides: "POST /api/documents/upload-url + POST /api/documents/{id}/confirm endpoints; old /upload removed"
      contains: "/upload-url"
    - path: "backend/api/auth.py"
      provides: "GET /api/auth/me/quota endpoint"
      contains: "/me/quota"
    - path: "backend/services/storage.py"
      provides: "delete_document decrements quota atomically; save_upload removed"
      contains: "GREATEST(0, used_bytes"
    - path: "backend/tasks/document_tasks.py"
      provides: "cleanup_abandoned_uploads Celery task + _cleanup_abandoned async body"
      contains: "cleanup_abandoned_uploads"
    - path: "backend/celery_app.py"
      provides: "beat_schedule entry for cleanup_abandoned_uploads every 30 minutes"
      contains: "beat_schedule"
    - path: "docker-compose.yml"
      provides: "MINIO_API_CORS_ALLOW_ORIGIN + MINIO_PUBLIC_ENDPOINT env vars; celery-beat service"
      contains: "MINIO_API_CORS_ALLOW_ORIGIN"
  key_links:
    - from: "backend/api/documents.py"
      to: "backend/storage/minio_backend.py"
      via: "backend.generate_presigned_put_url(object_key) and backend.stat_object(object_key)"
      pattern: "generate_presigned_put_url|stat_object"
    - from: "backend/api/documents.py"
      to: "quotas table"
      via: "atomic UPDATE quotas SET used_bytes = used_bytes + :delta WHERE (used_bytes + :delta) <= limit_bytes RETURNING used_bytes"
      pattern: "UPDATE quotas"
    - from: "backend/api/documents.py"
      to: "Celery (tasks.document_tasks.extract_and_classify)"
      via: "extract_and_classify.delay(str(doc.id)) after successful confirm"
      pattern: "extract_and_classify\\.delay"
    - from: "backend/celery_app.py"
      to: "backend/tasks/document_tasks.py"
      via: "beat_schedule: tasks.document_tasks.cleanup_abandoned_uploads every 30m"
      pattern: "cleanup_abandoned_uploads"
---

<objective>
Implement the presigned upload backend per CONTEXT.md D-04..D-07: replace multipart POST /upload with a 3-step flow (upload-url → browser PUT direct to MinIO → confirm) using the atomic quota UPDATE pattern from CLAUDE.md. Add GET /api/auth/me/quota for the sidebar quota bar. Wire the abandoned-upload cleanup Celery beat task per D-06.

Purpose: This plan is the only place the atomic quota SQL lives. Plan 03-03's auth guards depend on the endpoint surface defined here. Plan 03-05's frontend depends on the API contract defined here.
Output: 6 backend code changes + 1 docker-compose change. After this plan, an authenticated user can perform the full presigned upload/delete/quota cycle from curl; auth guards still allow anonymous access (closed in Plan 03-03).
</objective>

<execution_context>
@/Users/nik/.claude/get-shit-done/workflows/execute-plan.md
@/Users/nik/.claude/get-shit-done/templates/summary.md
</execution_context>

<context>
@.planning/ROADMAP.md
@.planning/STATE.md
@.planning/phases/03-document-migration-multi-user-isolation/03-CONTEXT.md
@.planning/phases/03-document-migration-multi-user-isolation/03-RESEARCH.md
@.planning/phases/03-document-migration-multi-user-isolation/03-PATTERNS.md
@.planning/phases/03-document-migration-multi-user-isolation/03-UI-SPEC.md
@.planning/phases/03-document-migration-multi-user-isolation/03-VALIDATION.md
@.planning/phases/03-document-migration-multi-user-isolation/03-01-SUMMARY.md
@CLAUDE.md

@backend/storage/base.py
@backend/storage/minio_backend.py
@backend/storage/__init__.py
@backend/api/documents.py
@backend/api/auth.py
@backend/services/storage.py
@backend/tasks/document_tasks.py
@backend/celery_app.py
@backend/config.py
@docker-compose.yml

<interfaces>
<!-- Contracts the executor needs without re-reading the codebase. -->

From backend/storage/base.py (current — Phase 1):
```
class StorageBackend(ABC):
    @abstractmethod
    async def put_object(self, user_id, document_id, file_bytes, extension, content_type) -> str
    @abstractmethod
    async def get_object(self, object_key) -> bytes
    @abstractmethod
    async def delete_object(self, object_key) -> None
    @abstractmethod
    async def presigned_get_url(self, object_key, expires_minutes=60) -> str
    @abstractmethod
    async def health_check(self) -> bool
```

From backend/storage/minio_backend.py (current):
```
class MinIOBackend(StorageBackend):
    def __init__(self, endpoint, access_key, secret_key, bucket, secure=False) -> None
    # Each method wraps the synchronous Minio SDK in asyncio.to_thread.
```

From CONTEXT.md decisions (D-05):
  POST /api/documents/upload-url body: {"filename": str, "content_type": str}
  Returns: {"upload_url": str, "document_id": str}
  POST /api/documents/{id}/confirm body: empty
  Returns: {"id": str, "size_bytes": int, "used_bytes": int, "status": "uploaded"}
  Quota exceeded: HTTP 413 detail = {"used_bytes": int, "limit_bytes": int, "rejected_bytes": int}

From backend/services/auth.py (Phase 2):
  def create_access_token(user_id: str, role: str) -> str

From backend/db/models.py:
  Document.user_id  — Plan 03-01 migration flips to NOT NULL
  Document.status   — 'pending' | 'uploaded' | 'classified' | 'classification_failed'
  Document.object_key
  Document.size_bytes
  Quota.limit_bytes — default 104857600 (100 MB)
  Quota.used_bytes

From backend/celery_app.py (current):
  celery_app.conf.task_routes = {
      "tasks.document_tasks.*": {"queue": "documents"},
      "tasks.email_tasks.*": {"queue": "email"},
  }
</interfaces>
</context>

<threat_model>
## Trust Boundaries

| Boundary | Description |
|----------|-------------|
| browser → MinIO (direct PUT) | Untrusted browser uses time-limited presigned URL; MinIO authenticates via HMAC signature; CORS preflight must succeed |
| browser → FastAPI /confirm | Authenticated user provides only document_id; FastAPI reads size_bytes from MinIO stat (never from client) |
| FastAPI /confirm → quotas table | Concurrent /confirm calls race against the same Quota row; PostgreSQL row-level lock + atomic UPDATE WHERE clause prevents overflow |
| Celery beat → DB+MinIO | Runs as docuvault_app; deletes its own rows (status='pending', age > 1h) and own MinIO objects only |

## STRIDE Threat Register

| Threat ID | Category | Component | Disposition | Mitigation Plan |
|-----------|----------|-----------|-------------|-----------------|
| T-03-04 | Spoofing | POST /api/documents/upload-url body filename/content_type | mitigate | object_key is computed server-side as {user_id}/{document_id}/{uuid4()}{ext} (STORE-02) — filename stored in DB column only; extension derived from Path(body.filename).suffix.lower() |
| T-03-05 | Tampering | Confirm endpoint quota delta | mitigate | size_bytes always comes from backend.stat_object(doc.object_key) (MinIO authoritative) — never from client body or request param |
| T-03-06 | Denial of Service | Concurrent /confirm uploads at quota boundary (SC2) | mitigate | Atomic SQL UPDATE quotas SET used_bytes = used_bytes + :delta WHERE (used_bytes + :delta) <= limit_bytes RETURNING used_bytes; fetchone() None → HTTP 413 (RESEARCH.md Finding 4 + Risk 2) |
| T-03-07 | Information Disclosure | Presigned URL leakage in logs | accept | 15-min TTL, single object key; leak risk acceptable for v1; do not log full URL — log only document_id |
| T-03-08 | Repudiation | Abandoned uploads pile up MinIO orphans | mitigate | Celery beat cleanup_abandoned_uploads every 30m deletes pending docs older than 1 hour and their MinIO objects (D-06) |
| T-03-09 | Information Disclosure | Browser→MinIO PUT CORS misconfiguration leaks origin | mitigate | MINIO_API_CORS_ALLOW_ORIGIN env var explicitly set to ${CORS_ORIGINS} (or http://localhost:5173 default) — not wildcard |
| T-03-10 | Tampering | Docker hostname in presigned URL (minio:9000 not browser-resolvable) | mitigate | Dual MinIO client: self._client (internal endpoint, used for stat/get/put/delete) and self._public_client (public endpoint, used for generate_presigned_put_url) per RESEARCH.md Finding 3 |
| T-03-SC | Tampering | pip installs | mitigate | No new package installs (minio + sqlalchemy already pinned in Phase 1 requirements.txt) |
</threat_model>

<tasks>

<task type="auto" tdd="true">
  <name>Task 1: Extend StorageBackend ABC + MinIOBackend with dual client, presigned PUT, stat_object; add config knobs and docker-compose env</name>
  <files>backend/storage/base.py, backend/storage/minio_backend.py, backend/storage/__init__.py, backend/config.py, docker-compose.yml</files>
  <read_first>
    - backend/storage/base.py — existing ABC method signatures and docstring style
    - backend/storage/minio_backend.py — existing constructor, asyncio.to_thread pattern, presigned_get_url example
    - backend/storage/__init__.py — current get_storage_backend factory
    - backend/config.py — Settings class shape, SettingsConfigDict usage, existing optional-string fields
    - docker-compose.yml — current minio service environment block, current celery-worker shape (model for celery-beat)
    - .planning/phases/03-document-migration-multi-user-isolation/03-RESEARCH.md — Finding 2 (presigned_put_object signature), Finding 3 (dual-client + CORS), Finding 5 (stat_object), Finding 10 (celery-beat service block)
  </read_first>
  <behavior>
    - StorageBackend ABC gains two new abstract methods: generate_presigned_put_url(object_key, expires_minutes=15) -> str and stat_object(object_key) -> int
    - MinIOBackend constructor accepts optional public_endpoint: str | None = None; stores self._public_client (Minio instance pointing at public_endpoint or falling back to endpoint)
    - MinIOBackend.generate_presigned_put_url uses self._public_client.presigned_put_object via asyncio.to_thread with timedelta(minutes=expires_minutes)
    - MinIOBackend.stat_object uses self._client.stat_object via asyncio.to_thread and returns .size (int)
    - get_storage_backend() reads settings.minio_public_endpoint and passes it to MinIOBackend
    - config.py adds minio_public_endpoint: str = "" (empty falls back to minio_endpoint inside MinIOBackend)
    - docker-compose.yml minio service env adds MINIO_API_CORS_ALLOW_ORIGIN: ${CORS_ORIGINS:-http://localhost:5173}; backend service env adds MINIO_PUBLIC_ENDPOINT; new celery-beat service runs celery -A celery_app beat --loglevel=info
  </behavior>
  <action>
    Modify `backend/storage/base.py`: append two `@abstractmethod async def` blocks after `presigned_get_url`, named `generate_presigned_put_url(self, object_key: str, expires_minutes: int = 15) -> str` and `stat_object(self, object_key: str) -> int`. Docstrings cite RESEARCH.md Finding 3 (public client requirement) and Finding 5 (returns authoritative size).

    Modify `backend/storage/minio_backend.py`: extend `__init__` signature with `public_endpoint: str | None = None`; after the existing `self._client = Minio(...)` block add `self._public_client = Minio(endpoint=(public_endpoint or endpoint), access_key=access_key, secret_key=secret_key, secure=secure)`. Append two new async methods (after `presigned_get_url`):
    - `generate_presigned_put_url(self, object_key, expires_minutes=15)` — `return await asyncio.to_thread(self._public_client.presigned_put_object, self._bucket, object_key, timedelta(minutes=expires_minutes))`
    - `stat_object(self, object_key)` — `result = await asyncio.to_thread(self._client.stat_object, self._bucket, object_key); return result.size`

    Modify `backend/storage/__init__.py`: extend `get_storage_backend()` to pass `public_endpoint=settings.minio_public_endpoint` to the `MinIOBackend(...)` constructor.

    Modify `backend/config.py`: add `minio_public_endpoint: str = ""` to the `Settings` class alongside the existing `minio_endpoint` field. Comment reference: `# RESEARCH.md Finding 3 — browser-resolvable hostname for presigned URLs`.

    Modify `docker-compose.yml`:
    1. In the `minio:` service `environment:` block, add `MINIO_API_CORS_ALLOW_ORIGIN: ${CORS_ORIGINS:-http://localhost:5173}` (RESEARCH.md Finding 3, T-03-09).
    2. In the `backend:` service `environment:` block, add `- MINIO_PUBLIC_ENDPOINT=${MINIO_PUBLIC_ENDPOINT:-localhost:9000}`.
    3. Append a new `celery-beat:` service mirroring `celery-worker` structure with `command: celery -A celery_app beat --loglevel=info` and the same `depends_on`, environment block, and build context (RESEARCH.md Finding 10).
  </action>
  <verify>
    <automated>cd backend && python -c "from storage.base import StorageBackend; assert 'generate_presigned_put_url' in dir(StorageBackend); assert 'stat_object' in dir(StorageBackend); from storage.minio_backend import MinIOBackend; import inspect; sig = inspect.signature(MinIOBackend.__init__); assert 'public_endpoint' in sig.parameters; print('OK')" && grep -c "MINIO_API_CORS_ALLOW_ORIGIN" docker-compose.yml && grep -c "celery-beat:" docker-compose.yml && grep -c "minio_public_endpoint" backend/config.py</automated>
  </verify>
  <done>
    StorageBackend ABC has 7 abstract methods (5 original + 2 new). MinIOBackend `__init__` accepts `public_endpoint`. docker-compose.yml contains `MINIO_API_CORS_ALLOW_ORIGIN` and `celery-beat:`. config.py contains `minio_public_endpoint`. All grep counts >= 1.
  </done>
</task>

<task type="auto" tdd="true">
  <name>Task 2: Implement upload-url, confirm, delete-with-quota, me/quota endpoints + remove old /upload + abandoned-upload Celery beat</name>
  <files>backend/api/documents.py, backend/api/auth.py, backend/services/storage.py, backend/tasks/document_tasks.py, backend/celery_app.py</files>
  <read_first>
    - backend/api/documents.py — existing handler structure (will be replaced)
    - backend/api/auth.py — existing /me endpoint pattern, router prefix, `from db.models import BackupCode, Quota, RefreshToken, User` import line
    - backend/services/storage.py — existing save_upload, delete_document, _backend() helper, _doc_to_dict, list_metadata
    - backend/tasks/document_tasks.py — existing extract_and_classify task; pattern for sync entry + async body
    - backend/tasks/email_tasks.py — _run_send_security_alert shape for async task body
    - backend/celery_app.py — current task_routes block (model for beat_schedule append)
    - backend/deps/auth.py — get_current_user signature (this plan does NOT wire auth into documents.py — Plan 03-03 does)
    - .planning/phases/03-document-migration-multi-user-isolation/03-RESEARCH.md — Finding 4 (atomic quota SQL), Finding 5 (stat_object error handling), Finding 9 (/me/quota), Finding 10 (Celery beat schedule)
    - .planning/phases/03-document-migration-multi-user-isolation/03-PATTERNS.md — atomic quota pattern, Celery task pattern
  </read_first>
  <behavior>
    - POST /api/documents/upload-url body {filename: str, content_type: str} returns {upload_url: str, document_id: str} and inserts a Document row with user_id=None (Wave 2 placeholder — Plan 03-03 replaces with current_user.id), status="pending", object_key=f"null-user/{doc_id}/{uuid4()}{ext}", size_bytes=0
    - POST /api/documents/{id}/confirm: load Document by id; call backend.stat_object(doc.object_key); catch minio.error.S3Error code="NoSuchKey" → return 422; on success set doc.size_bytes; execute atomic quota UPDATE; on row None → DELETE pending Document + remove MinIO object, return 413 {used_bytes, limit_bytes, rejected_bytes}; on success set status="uploaded", commit, enqueue extract_and_classify.delay(str(doc.id)); return {id, size_bytes, used_bytes, status}
    - GET /api/auth/me/quota: load Quota by current_user.id; return {used_bytes, limit_bytes}; if quota missing return 404
    - DELETE /api/documents/{id}: existing handler retained but services.storage.delete_document gains an atomic quota decrement (UPDATE quotas SET used_bytes = GREATEST(0, used_bytes - :delta) WHERE user_id = :uid) immediately before row delete
    - Old POST /api/documents/upload (multipart) is removed entirely; services.storage.save_upload is removed
    - tasks/document_tasks.py adds cleanup_abandoned_uploads task and async _cleanup_abandoned body that selects status="pending" AND created_at < now()-1h, deletes MinIO objects (try/except), then session.delete(doc), commits; returns {"cleaned": int}
    - celery_app.py adds beat_schedule entry "cleanup-abandoned-uploads" → "tasks.document_tasks.cleanup_abandoned_uploads" with schedule timedelta(minutes=30) and timezone "UTC"
    - test_upload_url_endpoint, test_confirm_endpoint, test_get_quota, test_quota_increment_atomic, test_concurrent_quota_race, test_quota_exceeded_response, test_delete_decrements_quota transition from xfail → pass (or xpass under strict=False)
  </behavior>
  <action>
    Rewrite `backend/api/documents.py`. Module-level imports: `from __future__ import annotations`, `import uuid`, `from pathlib import Path`, `from typing import Optional`, `from fastapi import APIRouter, Depends, HTTPException, Query, status`, `from pydantic import BaseModel`, `from sqlalchemy import text`, `from sqlalchemy.ext.asyncio import AsyncSession`, `from db.models import Document, Quota`, `from deps.db import get_db`, `from services import classifier, storage`, `from tasks.document_tasks import extract_and_classify`, `from storage import get_storage_backend`, `from minio.error import S3Error`. Router: `router = APIRouter(prefix="/api/documents", tags=["documents"])`.

    Define request model `class UploadUrlRequest(BaseModel): filename: str; content_type: str`. No response models — return plain dicts matching the documented shape.

    Endpoint POST `/upload-url`: handler signature `async def request_upload_url(body: UploadUrlRequest, session: AsyncSession = Depends(get_db))`. Generate `doc_id = uuid.uuid4()`; `suffix = Path(body.filename).suffix.lower()`; `object_key = f"null-user/{doc_id}/{uuid.uuid4()}{suffix}"` (Plan 03-03 replaces `"null-user"` with `str(current_user.id)`); insert `Document(id=doc_id, user_id=None, filename=body.filename, content_type=body.content_type, size_bytes=0, storage_backend="minio", status="pending", object_key=object_key)`; `session.add(doc); await session.commit()`; call `upload_url = await get_storage_backend().generate_presigned_put_url(object_key, expires_minutes=15)`; return `{"upload_url": upload_url, "document_id": str(doc_id)}`.

    Endpoint POST `/{doc_id}/confirm`: handler signature `async def confirm_upload(doc_id: str, session: AsyncSession = Depends(get_db))`. Parse `uid = uuid.UUID(doc_id)` (catch ValueError → 404); `doc = await session.get(Document, uid)`; if None → HTTPException(404, "Document not found"). Try `size = await get_storage_backend().stat_object(doc.object_key)`; catch `S3Error` where `getattr(e, "code", "") == "NoSuchKey"` → HTTPException(422, "Upload not found — presigned URL may have expired"). Set `doc.size_bytes = size`; `await session.flush()`. Conditional atomic quota update — if `doc.user_id is not None`: execute `text("UPDATE quotas SET used_bytes = used_bytes + :delta WHERE user_id = :uid AND (used_bytes + :delta) <= limit_bytes RETURNING used_bytes, limit_bytes")` with `{"delta": size, "uid": str(doc.user_id)}`; `row = result.fetchone()`. If `row is None` (quota exceeded): execute SELECT `text("SELECT used_bytes, limit_bytes FROM quotas WHERE user_id = :uid")`; `q = quota_row.fetchone()`; delete the pending Document via `await session.delete(doc)`; best-effort `try: await get_storage_backend().delete_object(doc.object_key) except Exception: pass`; `await session.commit()`; raise `HTTPException(status_code=413, detail={"used_bytes": q.used_bytes if q else 0, "limit_bytes": q.limit_bytes if q else 0, "rejected_bytes": size})`. On success path: `doc.status = "uploaded"`; `await session.commit()`; `extract_and_classify.delay(str(doc.id))`; return `{"id": str(doc.id), "size_bytes": size, "used_bytes": (row.used_bytes if row else 0), "status": "uploaded"}`. If `doc.user_id is None` (Wave 2 only): skip quota update entirely, set `doc.status = "uploaded"`, commit, enqueue, return `{"id": str(doc.id), "size_bytes": size, "used_bytes": 0, "status": "uploaded"}`.

    Endpoints GET `""` (list), GET `/{doc_id}` (get), DELETE `/{doc_id}` (delete), POST `/{doc_id}/classify` (reclassify): preserve the current handler bodies verbatim from existing `backend/api/documents.py` (lines 70-118) but rebound to the new router. NOTE: this plan does NOT yet add `Depends(get_current_user)` — Plan 03-03 adds auth wiring. Anonymous handlers remain in Wave 2.

    Modify `backend/services/storage.py`:
    - Delete `save_upload(session, file_bytes, original_name, mime_type)` function entirely (lines ~86-136) and its entry in `__all__`.
    - At top imports add `from sqlalchemy import text` (current import is `from sqlalchemy import select, delete` — extend the line).
    - Modify `delete_document(session, doc_id)`: after the existing `try: await _backend().delete_object(doc.object_key) ... except` block and BEFORE `await session.delete(doc)`, insert:
      ```
      if doc.user_id is not None:
          await session.execute(
              text("UPDATE quotas SET used_bytes = GREATEST(0, used_bytes - :delta) WHERE user_id = :uid"),
              {"delta": doc.size_bytes, "uid": str(doc.user_id)},
          )
      ```
      (Comment: "Atomic quota decrement (STORE-06, D-07). The user_id is None guard is removed in Plan 03-03.")

    Modify `backend/api/auth.py`: append a `GET /me/quota` handler after the existing `/me` handler. Handler:
    ```
    @router.get("/me/quota")
    async def get_my_quota(
        current_user: User = Depends(get_current_user),
        session: AsyncSession = Depends(get_db),
    ):
        q = await session.get(Quota, current_user.id)
        if q is None:
            raise HTTPException(status_code=404, detail="Quota not found")
        return {"used_bytes": q.used_bytes, "limit_bytes": q.limit_bytes}
    ```
    `Quota` is already imported on the existing `from db.models import BackupCode, Quota, RefreshToken, User` line — verify, no new import needed.

    Modify `backend/tasks/document_tasks.py`: append at end of file (after `_run`):
    ```
    @celery_app.task(name="tasks.document_tasks.cleanup_abandoned_uploads")
    def cleanup_abandoned_uploads() -> dict:
        """Periodic Celery beat task — deletes Document rows with status='pending'
        older than 1 hour and their MinIO objects (D-06).
        """
        return asyncio.run(_cleanup_abandoned())


    async def _cleanup_abandoned() -> dict:
        from datetime import datetime, timezone, timedelta
        from sqlalchemy import select

        from db.session import AsyncSessionLocal
        from db.models import Document
        from storage import get_storage_backend

        cutoff = datetime.now(timezone.utc) - timedelta(hours=1)
        async with AsyncSessionLocal() as session:
            result = await session.execute(
                select(Document).where(
                    Document.status == "pending",
                    Document.created_at < cutoff,
                )
            )
            docs = result.scalars().all()
            backend = get_storage_backend()
            cleaned = 0
            for doc in docs:
                try:
                    if doc.object_key:
                        await backend.delete_object(doc.object_key)
                except Exception:
                    pass  # MinIO object may not exist yet — safe to ignore
                await session.delete(doc)
                cleaned += 1
            await session.commit()
        return {"cleaned": cleaned}
    ```

    Modify `backend/celery_app.py`: at module top add `from datetime import timedelta as _timedelta`. After the existing `celery_app.conf.task_routes = {...}` block (before `autodiscover_tasks`), append:
    ```
    celery_app.conf.beat_schedule = {
        "cleanup-abandoned-uploads": {
            "task": "tasks.document_tasks.cleanup_abandoned_uploads",
            "schedule": _timedelta(minutes=30),
        },
    }
    celery_app.conf.timezone = "UTC"
    ```
  </action>
  <verify>
    <automated>cd backend && pytest tests/test_documents.py::test_upload_url_endpoint tests/test_documents.py::test_confirm_endpoint tests/test_documents.py::test_get_quota tests/test_quota.py -x -q 2>&1 | tail -30 && grep -c "generate_presigned_put_url" backend/api/documents.py && grep -c "stat_object" backend/api/documents.py && grep -c "UPDATE quotas" backend/api/documents.py && grep -v '^[[:space:]]*#' backend/services/storage.py | grep -c "GREATEST(0, used_bytes" && grep -c "cleanup_abandoned_uploads" backend/tasks/document_tasks.py && grep -c "beat_schedule" backend/celery_app.py && grep -c "/me/quota" backend/api/auth.py</automated>
  </verify>
  <done>
    All 7 listed pytest IDs pass (or xpass under strict=False). `backend/api/documents.py` contains `/upload-url`, `/{doc_id}/confirm`, calls `generate_presigned_put_url` + `stat_object` + atomic `UPDATE quotas`. `backend/services/storage.py` contains `GREATEST(0, used_bytes`. `backend/api/auth.py` contains `/me/quota`. `backend/tasks/document_tasks.py` contains `cleanup_abandoned_uploads`. `backend/celery_app.py` contains `beat_schedule`. All grep counts >= 1.
  </done>
</task>

</tasks>

<verification>
- Atomic quota race test green: `cd backend && pytest tests/test_quota.py::test_concurrent_quota_race -x -q` (SC2 for Phase 3)
- Quota rejection shape green: `cd backend && pytest tests/test_quota.py::test_quota_exceeded_response -x -q`
- Quota decrement on delete green: `cd backend && pytest tests/test_quota.py::test_delete_decrements_quota -x -q`
- Presigned upload endpoints green: `cd backend && pytest tests/test_documents.py::test_upload_url_endpoint tests/test_documents.py::test_confirm_endpoint -x -q`
- Quota read endpoint green: `cd backend && pytest tests/test_documents.py::test_get_quota -x -q`
- No legacy /upload route remains: `cd backend && grep -c '"/upload"' backend/api/documents.py` returns 0
</verification>

<success_criteria>
- StorageBackend ABC has `generate_presigned_put_url` and `stat_object` abstract methods
- MinIOBackend has dual MinIO client instances; presigned URL uses public client
- POST /api/documents/upload-url, POST /api/documents/{id}/confirm, GET /api/auth/me/quota exist and respond as documented
- DELETE /api/documents/{id} decrements quota atomically when doc.user_id is not None
- POST /api/documents/upload (multipart legacy) is removed; services.storage.save_upload is removed
- Celery beat schedule includes cleanup-abandoned-uploads every 30 minutes
- docker-compose.yml has MINIO_API_CORS_ALLOW_ORIGIN on minio service, MINIO_PUBLIC_ENDPOINT on backend service, and celery-beat service
- All Plan 03-01 stubs for STORE-03 / STORE-04 / STORE-05 / STORE-06 / D-05 pass or xpass
</success_criteria>

<output>
Create `.planning/phases/03-document-migration-multi-user-isolation/03-02-SUMMARY.md` when done — include the exact request/response shapes for /upload-url and /confirm; note the temporary `doc.user_id is None` guard that Plan 03-03 must remove.
</output>