kite/.planning/phases/03-document-migration-multi-user-isolation/03-03-PLAN.md

---
phase: 03-document-migration-multi-user-isolation
plan: 03
type: execute
wave: 3
depends_on:
  - 03-02
files_modified:
  - backend/deps/auth.py
  - backend/api/documents.py
  - backend/api/topics.py
  - backend/api/admin.py
  - backend/services/storage.py
  - backend/services/classifier.py
autonomous: true
requirements:
  - SEC-04
  - DOC-04

must_haves:
  truths:
    - "Every /api/documents/* handler requires a valid Bearer JWT and a non-admin user role; admins receive 403 on every document endpoint"
    - "Cross-user access to a document (by manipulating the doc_id path parameter) returns 404 — never reveals that the document exists"
    - "Newly uploaded Document rows have user_id = current_user.id; object_key embeds str(current_user.id) (no more 'null-user' sentinel)"
    - "GET /api/topics returns the union of system topics (user_id IS NULL) and the requesting user's own topics — never another user's topics"
    - "POST /api/admin/topics requires admin role; creates a Topic with user_id = NULL"
    - "Regular POST /api/topics creates a Topic with user_id = current_user.id"
    - "Atomic quota path in /confirm runs unconditionally (no more None-user fallback) for every authenticated upload"
  artifacts:
    - path: "backend/deps/auth.py"
      provides: "get_regular_user dependency (current_user with admin-403)"
      exports:
        - "get_regular_user"
    - path: "backend/api/documents.py"
      provides: "All handlers wired with get_regular_user dep + ownership assertion (doc.user_id == current_user.id else 404)"
      contains: "get_regular_user"
    - path: "backend/api/topics.py"
      provides: "All handlers wired with get_current_user dep + namespace-scoped queries"
      contains: "or_"
    - path: "backend/api/admin.py"
      provides: "POST /api/admin/topics endpoint (admin-only system topic creation)"
      contains: "/admin/topics"
    - path: "backend/services/storage.py"
      provides: "load_topics_for_user(session, user_id) + create_topic accepts user_id"
      exports:
        - "load_topics_for_user"
    - path: "backend/services/classifier.py"
      provides: "classify_document passes user_id through topic lookup and create_topic calls"
      contains: "load_topics_for_user"
  key_links:
    - from: "backend/api/documents.py"
      to: "backend/deps/auth.py"
      via: "current_user: User = Depends(get_regular_user) on every handler"
      pattern: "Depends\\(get_regular_user\\)"
    - from: "backend/api/documents.py"
      to: "documents table"
      via: "ownership assertion: if doc is None or doc.user_id != current_user.id → 404"
      pattern: "doc\\.user_id != current_user\\.id"
    - from: "backend/api/topics.py"
      to: "topics table"
      via: "WHERE (Topic.user_id == current_user.id OR Topic.user_id IS NULL)"
      pattern: "or_\\(Topic\\.user_id"
    - from: "backend/services/classifier.py"
      to: "backend/services/storage.py"
      via: "create_topic(session, name, user_id=doc.user_id) to keep AI-suggested topics in the user namespace"
      pattern: "create_topic.*user_id"
---

<objective>
Enforce per-user isolation across all `/api/documents/*` and `/api/topics/*` endpoints. Add the `get_regular_user` dependency (rejects admin), wire it into every documents handler with ownership assertions (404 on cross-user access per D-16), add `get_current_user` to topics with namespace-scoped queries (D-17), and create the admin-only `POST /api/admin/topics` endpoint (D-09). Remove the Wave 2 `null-user` placeholder from upload-url so newly created Documents carry the authenticated user_id.

Purpose: Closes Phase 2 D-07 / Phase 3 SC4 — admins return 403, cross-user returns 404. The atomic quota path from Plan 03-02 finally runs for real authenticated traffic.
Output: 6 file modifications. After this plan, the backend enforces full multi-user isolation for documents and topics; only Plan 03-04 (settings retirement + per-user AI) remains for the backend.
</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-VALIDATION.md
@.planning/phases/03-document-migration-multi-user-isolation/03-02-SUMMARY.md
@CLAUDE.md

@backend/deps/auth.py
@backend/api/documents.py
@backend/api/topics.py
@backend/api/admin.py
@backend/services/storage.py
@backend/services/classifier.py

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

From backend/deps/auth.py (Phase 2):
  async def get_current_user(credentials, session) -> User    # 401 on invalid/expired
  async def get_current_admin(user = Depends(get_current_user)) -> User   # 403 on non-admin

This plan adds:
  async def get_regular_user(user = Depends(get_current_user)) -> User    # 403 on admin

From backend/db/models.py:
  Document.user_id          # NOT NULL after Plan 03-01 migration
  Topic.user_id             # nullable: NULL = system topic; UUID = per-user
  Topic UniqueConstraint("user_id", "name", name="uq_topics_user_name")

From backend/services/storage.py (current — Plan 03-02 has removed save_upload):
  async def create_topic(session, name, description="", color="#6366f1") -> dict
  async def load_topics(session) -> list[dict]
  async def list_metadata(session, topic=None) -> list[dict]
  async def get_metadata(session, doc_id) -> Optional[dict]
  async def delete_document(session, doc_id) -> bool  # already decrements quota for non-None user_id

This plan changes:
  create_topic gains user_id: uuid.UUID | None = None parameter (deduplication scoped to user namespace)
  NEW: load_topics_for_user(session, user_id) -> list[dict]

From backend/services/classifier.py (current — Plan 03-04 changes signature for AI provider; this plan only changes topic plumbing):
  async def classify_document(session, doc_id, topic_names=None) -> list[str]
    — uses storage.load_topics(session) for the topic list and storage.create_topic(session, name) for new suggestions
  async def suggest_topics_for_document(session, doc_id) -> list[str]
</interfaces>
</context>

<threat_model>
## Trust Boundaries

| Boundary | Description |
|----------|-------------|
| browser → /api/documents/* | Authenticated user supplies doc_id path param; ownership assertion gates every read/write |
| browser → /api/topics/* | Authenticated user supplies topic ops; namespace filter prevents cross-user topic enumeration |
| admin token → /api/documents/* | Admin role is explicitly rejected with 403 — admins cannot read document content per CLAUDE.md |

## STRIDE Threat Register

| Threat ID | Category | Component | Disposition | Mitigation Plan |
|-----------|----------|-----------|-------------|-----------------|
| T-03-11 | Information Disclosure | Cross-user document access via doc_id manipulation | mitigate | Every handler taking doc_id asserts `doc is None or doc.user_id != current_user.id` and raises 404 (not 403) per D-16 / SEC-04 — attacker cannot distinguish "not found" from "wrong owner" |
| T-03-12 | Elevation of Privilege | Admin reading user document content | mitigate | get_regular_user dependency rejects admin with 403; applied to every /api/documents/* handler; PATTERNS.md "Auth Guard Pattern" |
| T-03-13 | Information Disclosure | Cross-user topic enumeration | mitigate | All topic queries filter by `or_(Topic.user_id == current_user.id, Topic.user_id.is_(None))`; create_topic dedup scoped by user_id (RESEARCH.md Finding 6 Risk 5) |
| T-03-14 | Elevation of Privilege | Regular user creating system topic | mitigate | POST /api/admin/topics requires `Depends(get_current_admin)`; regular POST /api/topics forces user_id=current_user.id |
| T-03-15 | Tampering | object_key forged with another user's UUID prefix | mitigate | object_key is computed server-side using `str(current_user.id)` — never accepts user-supplied prefix; CLAUDE.md "Every document/folder endpoint asserts resource.user_id == current_user.id" |
| T-03-16 | Spoofing | Anonymous traffic to /api/documents/* after auth wiring | mitigate | HTTPBearer auto_error=True raises 403 when no Authorization header present; get_current_user raises 401 on invalid/expired token |
| T-03-SC | Tampering | pip installs | mitigate | No new package installs |
</threat_model>

<tasks>

<task type="auto" tdd="true">
  <name>Task 1: Add get_regular_user dependency; wire auth + ownership + real user_id into /api/documents/*</name>
  <files>backend/deps/auth.py, backend/api/documents.py, backend/services/storage.py</files>
  <read_first>
    - backend/deps/auth.py — current get_current_user, get_current_admin signatures + docstrings (model for get_regular_user)
    - backend/api/documents.py — current state after Plan 03-02 (upload-url, confirm, list, get, delete, classify); locate the `"null-user"` placeholder and the `if doc.user_id is not None:` Wave 2 fallback
    - backend/services/storage.py — current list_metadata signature (must accept user_id filter); delete_document; get_metadata
    - backend/db/models.py — Document.user_id (now NOT NULL post Plan 03-01)
    - .planning/phases/03-document-migration-multi-user-isolation/03-CONTEXT.md — D-16 (404 not 403, admin 403)
    - .planning/phases/03-document-migration-multi-user-isolation/03-PATTERNS.md — "Auth Guard Pattern", "Ownership Assertion (SEC-04)"
    - .planning/phases/03-document-migration-multi-user-isolation/03-RESEARCH.md — Finding 7 (404 vs 403)
  </read_first>
  <behavior>
    - get_regular_user is a FastAPI dependency that wraps get_current_user and raises 403 if user.role == "admin"
    - Every handler in api/documents.py adds `current_user: User = Depends(get_regular_user)` parameter
    - upload-url replaces `"null-user"` in object_key with `str(current_user.id)` and sets `Document.user_id = current_user.id`
    - confirm endpoint removes the `if doc.user_id is not None:` Wave 2 fallback — the atomic quota path runs for every confirm
    - confirm endpoint adds ownership assertion before the stat_object call: `if doc is None or doc.user_id != current_user.id: raise HTTPException(404, "Document not found")`
    - list endpoint filters by user_id = current_user.id (services.storage.list_metadata accepts user_id param)
    - get endpoint asserts `doc.user_id == current_user.id` else 404
    - delete endpoint asserts ownership then calls services.storage.delete_document (which decrements quota — Plan 03-02 already wired this; remove the `if doc.user_id is not None:` guard there too)
    - classify endpoint asserts ownership
    - services.storage.list_metadata adds required `user_id: uuid.UUID` parameter and `Document.user_id == user_id` filter
    - services.storage.delete_document removes the `if doc.user_id is not None:` guard (now always non-None)
    - Tests test_cross_user_access_404, test_admin_cannot_access_documents, test_documents_require_auth transition from xfail → pass
  </behavior>
  <action>
    Modify `backend/deps/auth.py`: append a new dependency function:
    ```
    async def get_regular_user(user: User = Depends(get_current_user)) -> User:
        """Reject admin accounts on all /api/documents/* endpoints (D-16, SC4).

        Admin accounts cannot access document content (CLAUDE.md + SEC-04).
        Returns 403 (not 404) — the admin knows document endpoints exist.
        """
        if user.role == "admin":
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="Admin accounts cannot access document content",
            )
        return user
    ```

    Modify `backend/api/documents.py`:
    - Add imports `from db.models import User` and `from deps.auth import get_regular_user`.
    - In POST `/upload-url`: add `current_user: User = Depends(get_regular_user)` parameter; change `object_key = f"null-user/{doc_id}/...` to `object_key = f"{current_user.id}/{doc_id}/{uuid.uuid4()}{suffix}"`; change Document insert to `user_id=current_user.id`.
    - In POST `/{doc_id}/confirm`: add `current_user: User = Depends(get_regular_user)` parameter. After loading doc by id, add `if doc is None or doc.user_id != current_user.id: raise HTTPException(status_code=404, detail="Document not found")` (this replaces the simpler 404 check). Remove the `if doc.user_id is not None:` Wave 2 branch entirely — the atomic quota update path runs unconditionally. The else-branch that returned `used_bytes=0` for None-user is deleted.
    - In GET `""` (list): add `current_user: User = Depends(get_regular_user)` parameter; change `await storage.list_metadata(session, topic=topic)` to `await storage.list_metadata(session, user_id=current_user.id, topic=topic)`.
    - In GET `/{doc_id}`: add `current_user: User = Depends(get_regular_user)` parameter; replace the existing `meta = await storage.get_metadata(...)` flow with `doc = await session.get(Document, uuid.UUID(doc_id))`; check `if doc is None or doc.user_id != current_user.id: raise HTTPException(404, "Document not found")`; then call `meta = await storage.get_metadata(session, doc_id)` for the response shape (or replace with `_doc_to_dict` if more direct).
    - In DELETE `/{doc_id}`: add `current_user: User = Depends(get_regular_user)` parameter; load Document via `session.get(Document, uuid.UUID(doc_id))` and assert ownership before calling `storage.delete_document(session, doc_id)`; on `doc is None or doc.user_id != current_user.id` raise 404 (do not return success).
    - In POST `/{doc_id}/classify`: add `current_user: User = Depends(get_regular_user)` parameter; assert ownership via `doc = await session.get(Document, uuid.UUID(doc_id)); if doc is None or doc.user_id != current_user.id: raise HTTPException(404, "Document not found")` before calling classifier.

    Modify `backend/services/storage.py`:
    - Change `list_metadata` signature from `(session, topic=None)` to `(session, user_id: uuid.UUID, topic: Optional[str] = None)`. Update the `stmt = select(Document).order_by(...)` block to insert `.where(Document.user_id == user_id)` before any topic join.
    - Remove the `if doc.user_id is not None:` guard from `delete_document` added in Plan 03-02 — the quota decrement now always runs (post-migration, user_id is NOT NULL).
  </action>
  <verify>
    <automated>cd backend && pytest tests/test_documents.py::test_cross_user_access_404 tests/test_documents.py::test_admin_cannot_access_documents tests/test_documents.py::test_documents_require_auth tests/test_documents.py::test_upload_url_endpoint tests/test_documents.py::test_confirm_endpoint tests/test_quota.py -x -q 2>&1 | tail -20 && grep -c "get_regular_user" backend/deps/auth.py && grep -c "Depends(get_regular_user)" backend/api/documents.py && grep -v '^[[:space:]]*#' backend/api/documents.py | grep -c "doc.user_id != current_user.id" && grep -c "null-user" backend/api/documents.py</automated>
  </verify>
  <done>
    `get_regular_user` defined in deps/auth.py. `Depends(get_regular_user)` present in every handler of api/documents.py (>= 6 occurrences expected: upload-url, confirm, list, get, delete, classify). `null-user` literal NOT present in api/documents.py (count == 0). Ownership assertion `doc.user_id != current_user.id` appears in at least 4 handlers. All 6 listed tests pass or xpass.
  </done>
</task>

<task type="auto" tdd="true">
  <name>Task 2: Wire get_current_user + namespace filter into /api/topics/* + add POST /api/admin/topics + propagate user_id through classifier topic ops</name>
  <files>backend/api/topics.py, backend/api/admin.py, backend/services/storage.py, backend/services/classifier.py</files>
  <read_first>
    - backend/api/topics.py — current handler structure (will be modified)
    - backend/api/admin.py — pattern for `_admin: User = Depends(get_current_admin)` and Pydantic model definitions (UserCreate, AiConfigUpdate)
    - backend/services/storage.py — current `create_topic` (lines ~328-355), `load_topics` (lines ~289-295), `update_topic`, `delete_topic` signatures
    - backend/services/classifier.py — current `classify_document` body lines 33-58; `storage.load_topics` and `storage.create_topic` call sites
    - .planning/phases/03-document-migration-multi-user-isolation/03-CONTEXT.md — D-08, D-09, D-10, D-11, D-17
    - .planning/phases/03-document-migration-multi-user-isolation/03-PATTERNS.md — Topic namespace query (lines ~401-409)
    - .planning/phases/03-document-migration-multi-user-isolation/03-RESEARCH.md — Finding 6 (namespace query + dedup scoping + Risk 5)
  </read_first>
  <behavior>
    - GET /api/topics requires get_current_user; query filters `or_(Topic.user_id == current_user.id, Topic.user_id.is_(None))` via new service `load_topics_for_user`
    - POST /api/topics requires get_current_user; creates Topic with user_id = current_user.id (never NULL)
    - PATCH /api/topics/{id} requires get_current_user; the topic must belong to the user (user_id == current_user.id) — otherwise 404 (NOT 403 — same info-leak rationale as docs)
    - DELETE /api/topics/{id} requires get_current_user; same ownership rule. System topics (user_id IS NULL) cannot be edited or deleted via /api/topics/*
    - POST /api/topics/suggest requires get_current_user; existing classifier path
    - POST /api/admin/topics (NEW in api/admin.py) requires get_current_admin; creates Topic with user_id = NULL
    - services.storage.create_topic gains `user_id: uuid.UUID | None = None` parameter; dedup query is scoped: `WHERE lower(name) = :name AND user_id IS [NOT DISTINCT FROM] :user_id`
    - services.storage adds `load_topics_for_user(session, user_id) -> list[dict]` using the or_/is_ filter
    - services.classifier.classify_document replaces `storage.load_topics(session)` with `storage.load_topics_for_user(session, doc.user_id)` and passes `user_id=doc.user_id` into `storage.create_topic` for AI-suggested topics
    - Tests test_topic_namespace, test_admin_create_system_topic, test_regular_user_cannot_create_system_topic, test_topics_require_auth transition from xfail → pass
  </behavior>
  <action>
    Modify `backend/api/topics.py`. Add imports `from db.models import User` and `from deps.auth import get_current_user`. For every existing handler add `current_user: User = Depends(get_current_user)` as a parameter:

    - GET `""` (list_topics): replace `await storage.load_topics(session)` with `await storage.load_topics_for_user(session, user_id=current_user.id)`. Existing topic_doc_counts call: the doc count must also be scoped to current user — pass `user_id=current_user.id` into a new `storage.topic_doc_counts(session, user_id=current_user.id)` (add user filter to that helper as part of this task — add `Document.user_id == user_id` join condition).
    - POST `""` (create_topic): pass `user_id=current_user.id` into `storage.create_topic(...)`.
    - PATCH `/{topic_id}` (update_topic): before calling `storage.update_topic`, load the Topic and assert `topic.user_id == current_user.id` (no editing of system topics or other users' topics) — else 404. Implementation: `t = await session.get(Topic, uuid.UUID(topic_id)); if t is None or t.user_id != current_user.id: raise HTTPException(404, "Topic not found")`.
    - DELETE `/{topic_id}` (delete_topic): same ownership assertion before delete.
    - POST `/suggest` (suggest_topics): just add the dep; classifier handles the namespace.

    Modify `backend/api/admin.py`. Add a new Pydantic model `class SystemTopicCreate(BaseModel): name: str; description: str = ""; color: str = "#6366f1"`. Add a new endpoint:
    ```
    @router.post("/topics", status_code=status.HTTP_201_CREATED)
    async def create_system_topic(
        body: SystemTopicCreate,
        session: AsyncSession = Depends(get_db),
        _admin: User = Depends(get_current_admin),
    ) -> dict:
        """Admin creates a system topic visible to all users (D-09).

        System topics have user_id = NULL (per D-08 namespace model).
        """
        from services import storage
        topic = await storage.create_topic(
            session, body.name, body.description, body.color, user_id=None
        )
        return topic
    ```
    Import additions: add `from db.models import Topic` if not already imported (Topic is needed for any future admin topic edit/delete; this plan only adds POST).

    Modify `backend/services/storage.py`:
    1. Add `from sqlalchemy import or_` to the existing `from sqlalchemy import select, delete` import line.
    2. Add `import uuid` if not already at top (it is — verify).
    3. Modify `create_topic` signature: `async def create_topic(session, name, description="", color="#6366f1", user_id: uuid.UUID | None = None) -> dict`. Replace the dedup query with a namespace-scoped lookup. Because PostgreSQL needs `IS NOT DISTINCT FROM` semantics to compare NULLs but SQLite does not implement `IS NOT DISTINCT FROM`, use a branching approach:
       ```
       if user_id is None:
           q = await session.execute(
               select(Topic).where(
                   sql_func.lower(Topic.name) == name.lower(),
                   Topic.user_id.is_(None),
               )
           )
       else:
           q = await session.execute(
               select(Topic).where(
                   sql_func.lower(Topic.name) == name.lower(),
                   Topic.user_id == user_id,
               )
           )
       ```
       On insert, set `Topic(name=name, description=description, color=color, user_id=user_id)`.
    4. Add new function (after `load_topics`):
       ```
       async def load_topics_for_user(session, user_id: uuid.UUID) -> list:
           """Return system topics (user_id IS NULL) + user's own topics, ordered by name (D-17, DOC-04)."""
           q = await session.execute(
               select(Topic).where(
                   or_(Topic.user_id == user_id, Topic.user_id.is_(None))
               ).order_by(Topic.name)
           )
           return [
               {"id": str(t.id), "name": t.name, "description": t.description, "color": t.color}
               for t in q.scalars()
           ]
       ```
       Append `"load_topics_for_user"` to `__all__`.
    5. Modify `topic_doc_counts` to optionally scope by user: add `user_id: uuid.UUID | None = None` param; if non-None, join Document and filter by `Document.user_id == user_id` (so a user only sees the count of their own documents tagged to each topic).

    Modify `backend/services/classifier.py`:
    1. In `classify_document` (line ~39): replace `all_topics = await storage.load_topics(session)` with `all_topics = await storage.load_topics_for_user(session, user_id=doc.user_id)`. To access `doc.user_id`, fetch the Document at the top of the function: add `from db.models import Document; from sqlalchemy.ext.asyncio import AsyncSession; doc = await session.get(Document, uuid.UUID(doc_id))` (`uuid` is already imported via meta path; if not, add `import uuid`).
    2. Replace `await storage.create_topic(session, name.strip())` (line ~52) with `await storage.create_topic(session, name.strip(), user_id=doc.user_id)` to put AI-suggested topics in the user namespace (D-11).
    3. In `suggest_topics_for_document`: also replace `storage.load_topics(session)` if called (it is not, per current code) — only `provider.suggest_topics` is invoked, no change needed.
  </action>
  <verify>
    <automated>cd backend && pytest tests/test_topics.py::test_topic_namespace tests/test_topics.py::test_admin_create_system_topic tests/test_topics.py::test_regular_user_cannot_create_system_topic tests/test_topics.py::test_topics_require_auth tests/test_topics.py -x -q 2>&1 | tail -20 && grep -c "load_topics_for_user" backend/services/storage.py && grep -c "load_topics_for_user" backend/services/classifier.py && grep -c "or_(Topic.user_id" backend/services/storage.py && grep -c "/topics" backend/api/admin.py && grep -c "Depends(get_current_user)" backend/api/topics.py</automated>
  </verify>
  <done>
    All 4 topic-namespace tests pass or xpass. Existing topic tests (`test_list_topics_empty`, `test_create_topic`, etc.) still pass after auth wiring (they may need fixture updates — surface as a separate concern if they regress). `load_topics_for_user` is defined in storage.py and imported in classifier.py. `Depends(get_current_user)` appears in api/topics.py >= 5 times (list, create, update, delete, suggest). `POST /api/admin/topics` exists in api/admin.py.
  </done>
</task>

</tasks>

<verification>
- Cross-user access returns 404: `cd backend && pytest tests/test_documents.py::test_cross_user_access_404 -x -q`
- Admin gets 403 on documents: `cd backend && pytest tests/test_documents.py::test_admin_cannot_access_documents -x -q`
- Anonymous gets rejected: `cd backend && pytest tests/test_documents.py::test_documents_require_auth -x -q`
- Topic namespace isolation: `cd backend && pytest tests/test_topics.py::test_topic_namespace -x -q`
- Admin-only system topic creation: `cd backend && pytest tests/test_topics.py::test_admin_create_system_topic tests/test_topics.py::test_regular_user_cannot_create_system_topic -x -q`
- No null-user sentinel remains in api/documents.py: `cd backend && grep -c "null-user" backend/api/documents.py` returns 0
- Atomic quota path unconditional: `cd backend && grep -v '^[[:space:]]*#' backend/api/documents.py | grep -c "if doc.user_id is not None" | head -1` returns 0
</verification>

<success_criteria>
- get_regular_user dependency defined in deps/auth.py
- Every /api/documents/* handler injects Depends(get_regular_user) and asserts ownership (404 on cross-user)
- Every /api/topics/* handler injects Depends(get_current_user); queries are namespace-scoped
- POST /api/admin/topics exists and requires admin
- services.storage.create_topic accepts user_id; dedup scoped by namespace
- services.storage.load_topics_for_user implemented; consumed by classifier
- classify_document creates AI-suggested topics in the user namespace
- Phase 2 SC5 deferred item (admin JWT → 403 on documents) is fulfilled
</success_criteria>

<output>
Create `.planning/phases/03-document-migration-multi-user-isolation/03-03-SUMMARY.md` when done — list every endpoint now requiring auth, the ownership-assertion pattern (404 not 403), and any existing topic/document tests that needed fixture updates to handle the new auth requirement.
</output>