diff --git a/backend/api/admin/__init__.py b/backend/api/admin/__init__.py
new file mode 100644
index 0000000..f122afe
--- /dev/null
+++ b/backend/api/admin/__init__.py
@@ -0,0 +1,23 @@
+"""Admin API package — router aggregator.
+
+This file does ONLY router aggregation (Pitfall 2 prevention — T-08-04-04).
+No helpers, no models, no logic belong here.
+
+The prefix="/api/admin" is carried here so sub-routers have NO prefix (D-04):
+  users_router = APIRouter()   -> /api/admin/users via parent
+  quotas_router = APIRouter()  -> /api/admin/users/{id}/quota via parent
+  ai_router = APIRouter()      -> /api/admin/ai-config via parent
+
+main.py continues to use:
+  from api.admin import router as admin_router
+  app.include_router(admin_router)
+"""
+from fastapi import APIRouter
+from api.admin.users import router as users_router
+from api.admin.quotas import router as quotas_router
+from api.admin.ai import router as ai_router
+
+router = APIRouter(prefix="/api/admin", tags=["admin"])
+router.include_router(users_router)
+router.include_router(quotas_router)
+router.include_router(ai_router)
diff --git a/backend/api/admin/ai.py b/backend/api/admin/ai.py
new file mode 100644
index 0000000..c356302
--- /dev/null
+++ b/backend/api/admin/ai.py
@@ -0,0 +1,369 @@
+"""Admin AI configuration endpoints.
+
+Handles: get_ai_config_models, test_ai_connection, get_ai_config, update_system_ai_config.
+
+All handlers require get_current_admin (SEC-07, T-08-04-01).
+Sub-router has NO prefix — parent __init__.py carries /api/admin (D-04).
+
+_ai_config_to_dict is local to this module (only ai.py uses it — not in shared.py).
+"""
+from __future__ import annotations
+
+from typing import Optional
+
+from fastapi import APIRouter, Depends, HTTPException, Request, status
+from pydantic import BaseModel, ConfigDict, field_validator
+from sqlalchemy import select, update
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from ai import get_provider
+from ai.provider_config import ProviderConfig, PROVIDER_DEFAULTS
+from db.models import SystemSettings, User
+from deps.auth import get_current_admin
+from deps.db import get_db
+from deps.utils import get_client_ip
+from services.ai_config import encrypt_api_key, load_provider_config_by_id, validate_provider_id
+from services.audit import write_audit_log
+
+router = APIRouter()  # NO prefix — parent __init__.py carries /api/admin (D-04)
+
+
+# ── Local helper (only ai.py uses this — not in shared.py) ───────────────────
+
+def _ai_config_to_dict(row: SystemSettings) -> dict:
+    """Return a safe subset of SystemSettings fields — explicitly excludes api_key_enc.
+
+    has_api_key is the ONLY indicator that a key is stored (T-07-01 mitigated).
+    The raw encrypted value and any decrypted plaintext are NEVER returned.
+    """
+    return {
+        "provider_id": row.provider_id,
+        "base_url": row.base_url,
+        "model_name": row.model_name,
+        "context_chars": row.context_chars,
+        "is_active": row.is_active,
+        "has_api_key": row.api_key_enc is not None,
+        "updated_at": row.updated_at.isoformat() if row.updated_at else None,
+    }
+
+
+# ── Request models ────────────────────────────────────────────────────────────
+
+class SystemAiConfigUpdate(BaseModel):
+    """Request model for PUT /api/admin/ai-config (system-level provider configuration).
+
+    Security: extra="forbid" prevents mass-assignment of unexpected fields (T-07-13).
+    provider_id is validated against PROVIDER_DEFAULTS keys (T-07-13).
+    api_key is write-only: when None the existing api_key_enc is left untouched,
+    when "" the api_key_enc is cleared, when a non-empty string it is encrypted.
+    """
+
+    model_config = ConfigDict(extra="forbid")
+
+    provider_id: str
+    api_key: Optional[str] = None
+    base_url: Optional[str] = None
+    model_name: Optional[str] = None
+    context_chars: Optional[int] = None
+    is_active: Optional[bool] = None
+
+    @field_validator("provider_id")
+    @classmethod
+    def provider_must_be_known(cls, v: str) -> str:
+        return validate_provider_id(v)
+
+
+class TestConnectionRequest(BaseModel):
+    """Request body for POST /api/admin/ai-config/test-connection.
+
+    Unsaved form values (api_key, base_url, model_name) override the DB row so
+    admins can verify credentials before saving.  All override fields are optional;
+    omitting them falls back to whatever is stored in system_settings.
+    """
+
+    model_config = ConfigDict(extra="forbid")
+
+    provider_id: str
+    api_key: Optional[str] = None      # If non-empty, used instead of stored api_key_enc
+    base_url: Optional[str] = None     # If non-None, overrides DB base_url
+    model_name: Optional[str] = None   # If non-empty, overrides DB model_name
+
+    @field_validator("provider_id")
+    @classmethod
+    def provider_must_be_known(cls, v: str) -> str:
+        return validate_provider_id(v)
+
+
+# ── Endpoints ─────────────────────────────────────────────────────────────────
+
+@router.get("/ai-config/models")
+async def get_ai_config_models(
+    provider_id: str,
+    session: AsyncSession = Depends(get_db),
+    _admin: User = Depends(get_current_admin),
+) -> dict:
+    """Return the list of model IDs available from a provider's API (D-08).
+
+    Calls the provider's standard GET /models endpoint using the stored
+    config (base_url + api_key from system_settings).  Always returns 200
+    with {"models": [...]} — never 5xx on provider failure (returns empty list).
+
+    Security: requires get_current_admin; provider_id from query param only;
+    decrypted api_key never appears in the response.
+    """
+    import httpx  # noqa: PLC0415 — local import keeps startup fast
+
+    config = await load_provider_config_by_id(session, provider_id)
+
+    # Resolve base_url: prefer DB row, fall back to PROVIDER_DEFAULTS
+    if config and config.base_url:
+        base_url = config.base_url.rstrip("/")
+    else:
+        base_url = (PROVIDER_DEFAULTS.get(provider_id, {}).get("base_url") or "").rstrip("/")
+
+    if not base_url:
+        return {"models": [], "provider_id": provider_id}
+
+    api_key = config.api_key if config else ""
+
+    # Build request headers — Anthropic uses x-api-key; all others use Bearer
+    if provider_id == "anthropic":
+        headers = {
+            "x-api-key": api_key,
+            "anthropic-version": "2023-06-01",
+        }
+        models_url = "https://api.anthropic.com/v1/models"
+    else:
+        headers = {"Authorization": f"Bearer {api_key}"} if api_key else {}
+        models_url = f"{base_url}/models"
+
+    try:
+        async with httpx.AsyncClient(timeout=8.0) as client:
+            resp = await client.get(models_url, headers=headers)
+            resp.raise_for_status()
+            data = resp.json()
+
+        # Standard OpenAI-compat shape: {"data": [{"id": "...", ...}, ...]}
+        # Anthropic shape: {"data": [{"id": "...", ...}, ...]}
+        # Ollama OpenAI-compat: same shape
+        raw_list = data.get("data") or data.get("models") or []
+        model_ids: list[str] = sorted(
+            {
+                item["id"] if isinstance(item, dict) else str(item)
+                for item in raw_list
+                if item
+            }
+        )
+        return {"models": model_ids, "provider_id": provider_id}
+    except Exception as exc:
+        return {"models": [], "provider_id": provider_id, "error": str(exc)[:120]}
+
+
+@router.post("/ai-config/test-connection")
+async def test_ai_connection(
+    body: TestConnectionRequest,
+    session: AsyncSession = Depends(get_db),
+    _admin: User = Depends(get_current_admin),
+) -> dict:
+    """Test connectivity for an AI provider, optionally with unsaved form values (D-08).
+
+    Loads the stored system_settings row for body.provider_id, then overlays any
+    non-empty values from the request body so admins can verify credentials before
+    saving them to the database.
+
+    Override priority (highest -> lowest):
+      1. body.api_key / base_url / model_name  (unsaved form values)
+      2. system_settings DB row               (previously saved config)
+      3. PROVIDER_DEFAULTS                    (built-in fallback)
+
+    Returns {"ok": true/false, "provider_id": str} — never raises 5xx for
+    provider-side failures; surfaces as ok=False so the UI shows a clear status.
+
+    Security: requires get_current_admin; api_key from body is used only for the
+    in-flight health_check() call and is never stored or logged.
+    """
+    provider_id = body.provider_id
+    stored = await load_provider_config_by_id(session, provider_id)
+    defaults = PROVIDER_DEFAULTS.get(provider_id, {})
+
+    # Resolve effective values: body overrides DB, DB overrides PROVIDER_DEFAULTS
+    effective_api_key = (
+        body.api_key
+        if body.api_key
+        else (stored.api_key if stored else "")
+    )
+    effective_base_url = (
+        body.base_url
+        if body.base_url is not None
+        else (stored.base_url if stored else defaults.get("base_url"))
+    )
+    effective_model = (
+        body.model_name
+        if body.model_name
+        else (stored.model if stored else defaults.get("model", ""))
+    )
+
+    effective_config = ProviderConfig(
+        provider_id=provider_id,
+        api_key=effective_api_key,
+        base_url=effective_base_url,
+        model=effective_model,
+    )
+
+    try:
+        provider = get_provider(effective_config)
+        ok = await provider.health_check()
+        return {"ok": ok, "provider_id": provider_id}
+    except Exception:
+        return {"ok": False, "provider_id": provider_id, "reason": "health_check_failed"}
+
+
+@router.get("/ai-config")
+async def get_ai_config(
+    session: AsyncSession = Depends(get_db),
+    _admin: User = Depends(get_current_admin),
+) -> dict:
+    """Return all AI provider configurations for the admin panel (D-08).
+
+    Includes DB rows for providers that have been saved AND synthesised stubs
+    for providers that only exist in PROVIDER_DEFAULTS (so the admin UI always
+    shows all 10 providers even before any have been configured).
+
+    Security invariant: api_key_enc is NEVER returned (T-07-01).
+    Use has_api_key (bool) as the only indicator that a key is stored.
+    """
+    result = await session.execute(select(SystemSettings))
+    db_rows = result.scalars().all()
+
+    # Build a lookup for DB rows
+    db_by_provider: dict[str, SystemSettings] = {r.provider_id: r for r in db_rows}
+
+    providers_out = []
+    for pid in PROVIDER_DEFAULTS:
+        if pid in db_by_provider:
+            providers_out.append(_ai_config_to_dict(db_by_provider[pid]))
+        else:
+            # Synthesise a stub entry for providers with no DB row yet
+            defaults = PROVIDER_DEFAULTS[pid]
+            providers_out.append({
+                "provider_id": pid,
+                "base_url": defaults.get("base_url"),
+                "model_name": defaults.get("model", ""),
+                "context_chars": defaults.get("context_chars", 8000),
+                "is_active": False,
+                "has_api_key": False,
+                "updated_at": None,
+            })
+
+    return {"providers": providers_out}
+
+
+@router.put("/ai-config")
+async def update_system_ai_config(
+    body: SystemAiConfigUpdate,
+    request: Request,
+    session: AsyncSession = Depends(get_db),
+    _admin: User = Depends(get_current_admin),
+) -> dict:
+    """Create or update a system-level AI provider configuration (D-08, D-15).
+
+    Upsert semantics: if no row exists for body.provider_id, one is created using
+    PROVIDER_DEFAULTS for any omitted fields.
+
+    API key handling (T-07-01 mitigated):
+      - body.api_key is None -> leave existing api_key_enc untouched
+      - body.api_key == "" -> clear api_key_enc (set to NULL)
+      - body.api_key is a non-empty string -> HKDF-encrypt and store
+
+    is_active=True handling (T-07-03 mitigated):
+      When body.is_active is True, a single atomic UPDATE flips all rows:
+      SET is_active = (provider_id = :target_id)
+      This guarantees COUNT(WHERE is_active) == 1 with no read-then-write race.
+
+    Audit log (T-07-14 mitigated):
+      metadata_ contains only provider_id + fields_changed list — never the
+      api_key value itself.
+    """
+    from config import settings as _settings  # noqa: PLC0415
+
+    # Load existing row or create a new one from PROVIDER_DEFAULTS
+    stmt = select(SystemSettings).where(SystemSettings.provider_id == body.provider_id)
+    result = await session.execute(stmt)
+    row = result.scalar_one_or_none()
+
+    is_new = row is None
+    if is_new:
+        defaults = PROVIDER_DEFAULTS[body.provider_id]
+        row = SystemSettings(
+            provider_id=body.provider_id,
+            model_name=defaults.get("model", ""),
+            context_chars=defaults.get("context_chars", 8000),
+            base_url=defaults.get("base_url"),
+            is_active=False,
+            api_key_enc=None,
+        )
+
+    # Track which fields the caller explicitly set (for audit log — never api_key value)
+    fields_changed: list[str] = []
+
+    # Apply provided fields
+    if body.api_key is not None:
+        fields_changed.append("api_key")
+        if body.api_key == "":
+            row.api_key_enc = None
+        else:
+            master_key_str = _settings.cloud_creds_key
+            master_key_bytes = (
+                master_key_str.encode("utf-8")
+                if isinstance(master_key_str, str)
+                else master_key_str
+            )
+            row.api_key_enc = encrypt_api_key(master_key_bytes, body.provider_id, body.api_key)
+
+    if body.base_url is not None:
+        row.base_url = body.base_url
+        fields_changed.append("base_url")
+
+    if body.model_name is not None:
+        row.model_name = body.model_name
+        fields_changed.append("model_name")
+
+    if body.context_chars is not None:
+        row.context_chars = body.context_chars
+        fields_changed.append("context_chars")
+
+    if body.is_active is not None:
+        fields_changed.append("is_active")
+
+    if is_new:
+        session.add(row)
+        await session.flush()  # ensure row has an id before UPDATE
+
+    # Atomic is_active flip: SET is_active = (provider_id = :target) on ALL rows.
+    # Single UPDATE statement prevents dual-active race condition (T-07-03).
+    if body.is_active is True:
+        await session.execute(
+            update(SystemSettings).values(
+                is_active=(SystemSettings.provider_id == body.provider_id)
+            )
+        )
+        # Reflect the flip on the in-memory row
+        row.is_active = True
+
+    _ip_addr = get_client_ip(request)
+    await write_audit_log(
+        session,
+        event_type="admin.ai_config_changed",
+        user_id=None,
+        actor_id=_admin.id,
+        resource_id=None,
+        ip_address=_ip_addr,
+        metadata_={"provider_id": body.provider_id, "fields_changed": fields_changed},
+    )
+
+    await session.commit()
+
+    # Reload to pick up DB-generated updated_at after commit
+    await session.refresh(row)
+
+    return _ai_config_to_dict(row)
diff --git a/backend/api/admin/quotas.py b/backend/api/admin/quotas.py
new file mode 100644
index 0000000..42bfff0
--- /dev/null
+++ b/backend/api/admin/quotas.py
@@ -0,0 +1,115 @@
+"""Admin quota management endpoints.
+
+Handles: get_user_quota, update_user_quota.
+
+All handlers require get_current_admin (SEC-07, T-08-04-01).
+Sub-router has NO prefix — parent __init__.py carries /api/admin (D-04).
+"""
+from __future__ import annotations
+
+import uuid
+
+from fastapi import APIRouter, Depends, HTTPException, Request, status
+from pydantic import BaseModel, field_validator
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from db.models import Quota, User
+from deps.auth import get_current_admin
+from deps.db import get_db
+from deps.utils import get_client_ip
+from services.audit import write_audit_log
+from api.admin.shared import _user_to_dict
+
+router = APIRouter()  # NO prefix — parent __init__.py carries /api/admin (D-04)
+
+
+# ── Request models ────────────────────────────────────────────────────────────
+
+class QuotaUpdate(BaseModel):
+    limit_bytes: int
+
+    @field_validator("limit_bytes")
+    @classmethod
+    def must_be_positive(cls, v: int) -> int:
+        if v <= 0:
+            raise ValueError("limit_bytes must be greater than 0")
+        return v
+
+
+# ── Endpoints ─────────────────────────────────────────────────────────────────
+
+@router.get("/users/{user_id}/quota")
+async def get_user_quota(
+    user_id: uuid.UUID,
+    session: AsyncSession = Depends(get_db),
+    _admin: User = Depends(get_current_admin),
+) -> dict:
+    """Return quota details for a user (ADMIN-04).
+
+    Quota info is admin-visible operational data — no PII, no document content
+    (T-02-31 disposition: accept).
+    """
+    quota = await session.get(Quota, user_id)
+    if quota is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Quota not found")
+
+    return {
+        "user_id": str(quota.user_id),
+        "limit_bytes": quota.limit_bytes,
+        "used_bytes": quota.used_bytes,
+        "limit_mb": quota.limit_bytes // 1048576,
+        "used_mb": quota.used_bytes // 1048576,
+    }
+
+
+@router.patch("/users/{user_id}/quota")
+async def update_user_quota(
+    user_id: uuid.UUID,
+    body: QuotaUpdate,
+    request: Request,
+    session: AsyncSession = Depends(get_db),
+    _admin: User = Depends(get_current_admin),
+) -> dict:
+    """Adjust a user's storage quota (ADMIN-04).
+
+    If the new limit is below current usage, still applies the change but
+    returns warning=True with an explanatory message. Uploads will be blocked
+    but existing documents are preserved.
+    """
+    quota = await session.get(Quota, user_id)
+    if quota is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Quota not found")
+
+    warning = body.limit_bytes < quota.used_bytes
+    warning_message = (
+        "New limit is below current usage. Uploads will be blocked but existing documents are preserved."
+        if warning
+        else None
+    )
+
+    _ip_addr = get_client_ip(request)
+    old_limit = quota.limit_bytes
+    quota.limit_bytes = body.limit_bytes
+    session.add(quota)
+
+    # D-13: quota changed event
+    await write_audit_log(
+        session,
+        event_type="admin.quota_changed",
+        user_id=user_id,
+        actor_id=_admin.id,
+        resource_id=None,
+        ip_address=_ip_addr,
+        metadata_={"old_bytes": old_limit, "new_bytes": body.limit_bytes},
+    )
+    await session.commit()
+
+    response: dict = {
+        "user_id": str(quota.user_id),
+        "limit_bytes": quota.limit_bytes,
+        "used_bytes": quota.used_bytes,
+        "warning": warning,
+    }
+    if warning_message:
+        response["message"] = warning_message
+    return response
diff --git a/backend/api/admin/shared.py b/backend/api/admin/shared.py
new file mode 100644
index 0000000..193f9e6
--- /dev/null
+++ b/backend/api/admin/shared.py
@@ -0,0 +1,28 @@
+"""Shared helpers for the admin API package.
+
+These helpers are used by 2+ admin sub-modules and must not live in __init__.py
+to avoid circular imports (T-08-04-04, RESEARCH.md Pitfall 2).
+
+_ai_config_to_dict lives in ai.py (only ai.py uses it — local is correct).
+"""
+from __future__ import annotations
+
+from db.models import User
+
+
+def _user_to_dict(user: User) -> dict:
+    """Return a safe subset of User fields — never includes password_hash,
+    credentials_enc, totp_secret, or any document content (T-02-27, SEC-07).
+    """
+    return {
+        "id": str(user.id),
+        "handle": user.handle,
+        "email": user.email,
+        "role": user.role,
+        "is_active": user.is_active,
+        "totp_enabled": user.totp_enabled,
+        "ai_provider": user.ai_provider,
+        "ai_model": user.ai_model,
+        "password_must_change": user.password_must_change,
+        "created_at": user.created_at.isoformat() if user.created_at else None,
+    }
diff --git a/backend/api/admin/users.py b/backend/api/admin/users.py
new file mode 100644
index 0000000..a7d833a
--- /dev/null
+++ b/backend/api/admin/users.py
@@ -0,0 +1,431 @@
+"""Admin user-management endpoints.
+
+Handles: list_users, create_user, update_user_status, initiate_password_reset,
+         update_ai_config (per-user), delete_user, create_system_topic.
+
+All handlers require get_current_admin (SEC-07, T-08-04-01).
+Sub-router has NO prefix — parent __init__.py carries /api/admin (D-04).
+"""
+from __future__ import annotations
+
+import time
+import uuid
+from typing import Optional
+
+from config import settings
+from fastapi import APIRouter, Depends, HTTPException, Request, status
+from pydantic import BaseModel, EmailStr, Field, field_validator
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from db.models import CloudConnection, Document, Quota, RefreshToken, Topic, User
+from deps.auth import get_current_admin
+from deps.db import get_db
+from deps.utils import get_client_ip
+from services.audit import write_audit_log
+from services.auth import hash_password, revoke_all_refresh_tokens, validate_password_strength, verify_password
+from storage import get_storage_backend, get_storage_backend_for_document
+from api.admin.shared import _user_to_dict
+
+router = APIRouter()  # NO prefix — parent __init__.py carries /api/admin (D-04)
+
+# ── Constants ─────────────────────────────────────────────────────────────────
+
+_DEFAULT_QUOTA_BYTES = 104857600  # 100 MB free-tier default (D-06)
+
+
+# ── Request models ────────────────────────────────────────────────────────────
+
+class UserCreate(BaseModel):
+    handle: str
+    email: EmailStr
+    password: str
+    role: str = "user"
+
+    @field_validator("password")
+    @classmethod
+    def password_strength(cls, v: str) -> str:
+        validate_password_strength(v)
+        return v
+
+
+class UserStatusUpdate(BaseModel):
+    is_active: bool
+
+
+class UserAiConfigUpdate(BaseModel):
+    ai_provider: Optional[str] = None
+    ai_model: Optional[str] = None
+
+
+class SystemTopicCreate(BaseModel):
+    """Request model for admin system topic creation (D-09)."""
+
+    name: str
+    description: str = ""
+    color: str = "#6366f1"
+
+
+class UserDeleteConfirm(BaseModel):
+    """Admin password confirmation required before hard-deleting a user (ADMIN-02, T-05-11-01)."""
+
+    admin_password: str = Field(..., min_length=1)
+
+
+# ── Endpoints ─────────────────────────────────────────────────────────────────
+
+@router.get("/users")
+async def list_users(
+    session: AsyncSession = Depends(get_db),
+    _admin: User = Depends(get_current_admin),
+) -> dict:
+    """List all users, ordered by created_at DESC.
+
+    Response shape: { items: [...safe user fields...] }
+    Never includes password_hash, credentials_enc, or document content (T-02-27).
+    """
+    result = await session.execute(
+        select(User).order_by(User.created_at.desc())
+    )
+    users = result.scalars().all()
+    return {"items": [_user_to_dict(u) for u in users]}
+
+
+@router.post("/users", status_code=status.HTTP_201_CREATED)
+async def create_user(
+    request: Request,
+    body: UserCreate,
+    session: AsyncSession = Depends(get_db),
+    _admin: User = Depends(get_current_admin),
+) -> dict:
+    """Admin creates a new user account (ADMIN-01).
+
+    - password_must_change=True forces the user to change their password on
+      first login (T-02-32, D-06).
+    - Quota row initialized at 100 MB (D-06).
+    - Returns 409 if email or handle is already taken.
+    """
+    # Check uniqueness
+    existing_email = await session.execute(
+        select(User).where(User.email == str(body.email))
+    )
+    if existing_email.scalar_one_or_none() is not None:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="Email already registered",
+        )
+
+    existing_handle = await session.execute(
+        select(User).where(User.handle == body.handle)
+    )
+    if existing_handle.scalar_one_or_none() is not None:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="Handle already taken",
+        )
+
+    new_user = User(
+        id=uuid.uuid4(),
+        handle=body.handle,
+        email=str(body.email),
+        password_hash=hash_password(body.password),
+        role=body.role,
+        is_active=True,
+        totp_enabled=False,
+        password_must_change=True,  # ADMIN-01: force password change on first login
+    )
+    session.add(new_user)
+
+    quota = Quota(
+        user_id=new_user.id,
+        limit_bytes=_DEFAULT_QUOTA_BYTES,
+        used_bytes=0,
+    )
+    session.add(quota)
+    await session.flush()  # persist User + Quota before audit_log FK references them
+    # D-13: admin user created event
+    _ip_addr = get_client_ip(request)
+    await write_audit_log(
+        session,
+        event_type="admin.user_created",
+        user_id=new_user.id,
+        actor_id=_admin.id,
+        resource_id=new_user.id,
+        ip_address=_ip_addr,
+    )
+    await session.commit()
+
+    return {
+        "id": str(new_user.id),
+        "handle": new_user.handle,
+        "email": new_user.email,
+        "role": new_user.role,
+        "created_at": new_user.created_at.isoformat() if new_user.created_at else None,
+    }
+
+
+@router.patch("/users/{user_id}/status")
+async def update_user_status(
+    user_id: uuid.UUID,
+    body: UserStatusUpdate,
+    request: Request,
+    session: AsyncSession = Depends(get_db),
+    _admin: User = Depends(get_current_admin),
+) -> dict:
+    """Deactivate or reactivate a user account (ADMIN-02).
+
+    - Prevents deactivating the last active admin (T-02-29).
+    - On deactivation: all refresh tokens are revoked (family revocation).
+    """
+    user = await session.get(User, user_id)
+    if user is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
+
+    # Guard: cannot deactivate the only remaining active admin (T-02-29)
+    if not body.is_active and user.role == "admin":
+        count_result = await session.execute(
+            select(func.count(User.id)).where(
+                User.role == "admin",
+                User.is_active.is_(True),
+            )
+        )
+        active_admin_count = count_result.scalar_one()
+        if active_admin_count <= 1:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Cannot deactivate the only admin",
+            )
+
+    _ip_addr = get_client_ip(request)
+    user.is_active = body.is_active
+
+    if not body.is_active:
+        # Revoke all refresh tokens on deactivation
+        await revoke_all_refresh_tokens(session, user.id)
+        # Revoke any pre-deactivation access tokens still within their TTL (T-7.2-01)
+        await request.app.state.redis.set(
+            f"user_nbf:{user.id}",
+            int(time.time()),
+            ex=settings.access_token_expire_minutes * 60,
+        )
+
+    session.add(user)
+
+    # D-13: user deactivated/activated event
+    _event = "admin.user_deactivated" if not body.is_active else "admin.user_activated"
+    await write_audit_log(
+        session,
+        event_type=_event,
+        user_id=user.id,
+        actor_id=_admin.id,
+        resource_id=user.id,
+        ip_address=_ip_addr,
+    )
+    await session.commit()
+
+    return {
+        "id": str(user.id),
+        "handle": user.handle,
+        "email": user.email,
+        "is_active": user.is_active,
+    }
+
+
+@router.post("/users/{user_id}/password-reset", status_code=status.HTTP_202_ACCEPTED)
+async def initiate_password_reset(
+    user_id: uuid.UUID,
+    session: AsyncSession = Depends(get_db),
+    _admin: User = Depends(get_current_admin),
+) -> dict:
+    """Admin initiates a password reset for a user (ADMIN-03).
+
+    Sends the reset email via Celery. Does NOT:
+      - return a reset token (T-02-30)
+      - grant admin access to the account
+      - log in as the target user (ADMIN-07 — no impersonation)
+
+    Returns 202 immediately regardless of email delivery status.
+    """
+    user = await session.get(User, user_id)
+    if user is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
+
+    from services.auth import create_password_reset_token  # noqa: PLC0415
+    from config import settings as _settings  # noqa: PLC0415
+
+    reset_token = create_password_reset_token(str(user.id))
+    reset_link = f"{_settings.frontend_url}/password-reset/confirm?token={reset_token}"
+
+    # Deferred import to avoid circular imports (same pattern as document_tasks)
+    from tasks.email_tasks import send_reset_email  # noqa: PLC0415
+    send_reset_email.delay(user.email, reset_link)
+
+    return {"message": "Password reset email sent"}
+
+
+@router.patch("/users/{user_id}/ai-config")
+async def update_ai_config(
+    user_id: uuid.UUID,
+    body: UserAiConfigUpdate,
+    request: Request,
+    session: AsyncSession = Depends(get_db),
+    _admin: User = Depends(get_current_admin),
+) -> dict:
+    """Assign AI provider and model for a user (ADMIN-05).
+
+    Users cannot change their own AI provider or model (PROJECT.md Key Decision).
+    Only admins have this capability.
+    """
+    user = await session.get(User, user_id)
+    if user is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
+
+    _ip_addr = get_client_ip(request)
+    user.ai_provider = body.ai_provider
+    user.ai_model = body.ai_model
+    session.add(user)
+
+    # D-13: AI provider assigned event
+    await write_audit_log(
+        session,
+        event_type="admin.ai_provider_assigned",
+        user_id=user_id,
+        actor_id=_admin.id,
+        resource_id=None,
+        ip_address=_ip_addr,
+        metadata_={"provider": body.ai_provider, "model": body.ai_model},
+    )
+    await session.commit()
+
+    return {
+        "id": str(user.id),
+        "email": user.email,
+        "ai_provider": user.ai_provider,
+        "ai_model": user.ai_model,
+    }
+
+
+@router.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
+async def delete_user(
+    user_id: uuid.UUID,
+    body: UserDeleteConfirm,
+    request: Request,
+    session: AsyncSession = Depends(get_db),
+    _admin: User = Depends(get_current_admin),
+) -> None:
+    """Delete a user account and clean up all their MinIO objects (SEC-09, D-19).
+
+    Security invariants:
+      - Admin password verified via Argon2 before any deletion (T-05-11-01)
+      - Cannot delete admin accounts (T-04-07-04)
+      - MinIO objects are deleted BEFORE DB records are removed (SEC-09)
+      - MinIO deletion is best-effort (try/except) — DB row is deleted regardless
+      - Audit log written with event_type="admin.user_deleted"
+    """
+    # T-05-11-01: Verify admin password before performing any destructive action.
+    # Fail fast — no DB reads for the target user until the admin is confirmed.
+    if not verify_password(body.admin_password, _admin.password_hash):
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Invalid admin password",
+        )
+
+    user = await session.get(User, user_id)
+    if user is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
+
+    # T-04-07-04: Cannot delete admin accounts
+    if user.role == "admin":
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Cannot delete admin accounts",
+        )
+
+    _ip_addr = get_client_ip(request)
+
+    # SEC-09 (cloud): purge cloud-stored documents and credentials BEFORE DB delete.
+    # Must run before MinIO cleanup so that credentials are still available to build
+    # the cloud backend instances for delete_object calls.
+    cloud_conns_result = await session.execute(
+        select(CloudConnection).where(CloudConnection.user_id == user_id)
+    )
+    cloud_conns = cloud_conns_result.scalars().all()
+    for conn in cloud_conns:
+        # Delete cloud objects stored in this provider for this user
+        cloud_docs_result = await session.execute(
+            select(Document).where(
+                Document.user_id == user_id,
+                Document.storage_backend == conn.provider,
+            )
+        )
+        for doc in cloud_docs_result.scalars().all():
+            try:
+                backend = await get_storage_backend_for_document(doc, user, session)
+                await backend.delete_object(doc.object_key)
+            except Exception:
+                pass  # Best-effort cloud object cleanup; deletion proceeds regardless
+        # Purge the credentials row (FK cascade would also remove it, but explicit
+        # deletion here guarantees credentials_enc is gone before commit — SEC-09)
+        await session.delete(conn)
+    if cloud_conns:
+        await session.flush()  # Flush connection deletes before user delete
+        await write_audit_log(
+            session,
+            event_type="cloud.credentials_purged",
+            user_id=user_id,
+            actor_id=_admin.id,
+            resource_id=user_id,
+            ip_address=_ip_addr,
+            metadata_={"providers": [c.provider for c in cloud_conns]},
+        )
+
+    # SEC-09 (minio): collect all user documents and delete MinIO objects BEFORE DB delete
+    docs_result = await session.execute(
+        select(Document).where(Document.user_id == user_id)
+    )
+    user_docs = docs_result.scalars().all()
+
+    storage = get_storage_backend()
+    for doc in user_docs:
+        try:
+            await storage.delete_object(doc.object_key)
+        except Exception:
+            pass  # Best-effort MinIO cleanup; DB deletion proceeds regardless
+
+    # D-13: audit log BEFORE deleting the user row (user FK still valid at flush time)
+    await write_audit_log(
+        session,
+        event_type="admin.user_deleted",
+        user_id=user_id,
+        actor_id=_admin.id,
+        resource_id=user_id,
+        ip_address=_ip_addr,
+    )
+    await session.flush()
+
+    # Delete user record (CASCADE removes quota, documents, refresh_tokens, etc.)
+    await session.delete(user)
+    await session.commit()
+
+
+@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:
+    """Create a system topic visible to all users (D-09, DOC-04).
+
+    System topics have user_id = NULL, making them visible to every user as
+    defaults in their topic namespace. Only admins can create system topics.
+    Regular users create per-user topics via POST /api/topics.
+
+    Deduplication: case-insensitive match within the system namespace (user_id IS NULL).
+    Returns the existing system topic if one with the same name already exists.
+    """
+    from services import storage  # noqa: PLC0415
+
+    topic = await storage.create_topic(
+        session, body.name, body.description, body.color, user_id=None
+    )
+    return topic
diff --git a/backend/api/admin_OLD_REMOVE_IN_TASK_3.py b/backend/api/admin_OLD_REMOVE_IN_TASK_3.py
new file mode 100644
index 0000000..ee26e7e
--- /dev/null
+++ b/backend/api/admin_OLD_REMOVE_IN_TASK_3.py
@@ -0,0 +1,934 @@
+"""
+Admin API endpoints for DocuVault.
+
+All handlers require get_current_admin (SEC-07, D-08) — no handler uses
+get_current_user alone.
+
+Implements:
+  GET    /api/admin/users                    — list all users (ADMIN-01)
+  POST   /api/admin/users                    — create user (ADMIN-01)
+  PATCH  /api/admin/users/{id}/status        — deactivate/reactivate (ADMIN-02)
+  POST   /api/admin/users/{id}/password-reset — initiate reset email (ADMIN-03)
+  GET    /api/admin/users/{id}/quota          — view quota (ADMIN-04)
+  PATCH  /api/admin/users/{id}/quota          — adjust quota (ADMIN-04)
+  PATCH  /api/admin/users/{id}/ai-config      — assign AI provider/model (ADMIN-05)
+
+Security invariants:
+  - Every handler injects Depends(get_current_admin) — verified by grep count
+  - _user_to_dict() whitelist helper prevents accidental field leakage (T-02-27)
+  - No impersonation endpoint — ADMIN-07 enforced by omission (T-02-28)
+  - Admin-created users: password_must_change=True (ADMIN-01, T-02-32)
+  - Deactivation of sole admin prevented (T-02-29)
+  - Password reset sends email via Celery; does not return token (T-02-30)
+"""
+from __future__ import annotations
+
+import time
+import uuid
+from datetime import datetime
+from typing import Optional
+
+from config import settings
+from fastapi import APIRouter, Depends, HTTPException, Request, status
+from pydantic import BaseModel, ConfigDict, EmailStr, Field, field_validator
+from sqlalchemy import func, select, update
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from ai import get_provider
+from ai.provider_config import ProviderConfig, PROVIDER_DEFAULTS
+from db.models import CloudConnection, Document, Quota, RefreshToken, SystemSettings, Topic, User
+from deps.auth import get_current_admin
+from deps.db import get_db
+from deps.utils import get_client_ip
+from services.ai_config import encrypt_api_key, load_provider_config_by_id
+from services.audit import write_audit_log
+from services.auth import hash_password, revoke_all_refresh_tokens, validate_password_strength, verify_password
+from storage import get_storage_backend, get_storage_backend_for_document
+
+router = APIRouter(prefix="/api/admin", tags=["admin"])
+
+# ── Constants ─────────────────────────────────────────────────────────────────
+
+_DEFAULT_QUOTA_BYTES = 104857600  # 100 MB free-tier default (D-06)
+
+
+
+# ── Safe response helpers ─────────────────────────────────────────────────────
+
+def _ai_config_to_dict(row: SystemSettings) -> dict:
+    """Return a safe subset of SystemSettings fields — explicitly excludes api_key_enc.
+
+    has_api_key is the ONLY indicator that a key is stored (T-07-01 mitigated).
+    The raw encrypted value and any decrypted plaintext are NEVER returned.
+    """
+    return {
+        "provider_id": row.provider_id,
+        "base_url": row.base_url,
+        "model_name": row.model_name,
+        "context_chars": row.context_chars,
+        "is_active": row.is_active,
+        "has_api_key": row.api_key_enc is not None,
+        "updated_at": row.updated_at.isoformat() if row.updated_at else None,
+    }
+
+
+def _user_to_dict(user: User) -> dict:
+    """Return a safe subset of User fields — never includes password_hash,
+    credentials_enc, totp_secret, or any document content (T-02-27, SEC-07).
+    """
+    return {
+        "id": str(user.id),
+        "handle": user.handle,
+        "email": user.email,
+        "role": user.role,
+        "is_active": user.is_active,
+        "totp_enabled": user.totp_enabled,
+        "ai_provider": user.ai_provider,
+        "ai_model": user.ai_model,
+        "password_must_change": user.password_must_change,
+        "created_at": user.created_at.isoformat() if user.created_at else None,
+    }
+
+
+
+# ── Request models ────────────────────────────────────────────────────────────
+
+class UserCreate(BaseModel):
+    handle: str
+    email: EmailStr
+    password: str
+    role: str = "user"
+
+    @field_validator("password")
+    @classmethod
+    def password_strength(cls, v: str) -> str:
+        validate_password_strength(v)
+        return v
+
+
+class UserStatusUpdate(BaseModel):
+    is_active: bool
+
+
+class QuotaUpdate(BaseModel):
+    limit_bytes: int
+
+    @field_validator("limit_bytes")
+    @classmethod
+    def must_be_positive(cls, v: int) -> int:
+        if v <= 0:
+            raise ValueError("limit_bytes must be greater than 0")
+        return v
+
+
+class UserAiConfigUpdate(BaseModel):
+    ai_provider: Optional[str] = None
+    ai_model: Optional[str] = None
+
+
+class SystemAiConfigUpdate(BaseModel):
+    """Request model for PUT /api/admin/ai-config (system-level provider configuration).
+
+    Security: extra="forbid" prevents mass-assignment of unexpected fields (T-07-13).
+    provider_id is validated against PROVIDER_DEFAULTS keys (T-07-13).
+    api_key is write-only: when None the existing api_key_enc is left untouched,
+    when "" the api_key_enc is cleared, when a non-empty string it is encrypted.
+    """
+
+    model_config = ConfigDict(extra="forbid")
+
+    provider_id: str
+    api_key: Optional[str] = None
+    base_url: Optional[str] = None
+    model_name: Optional[str] = None
+    context_chars: Optional[int] = None
+    is_active: Optional[bool] = None
+
+    @field_validator("provider_id")
+    @classmethod
+    def provider_must_be_known(cls, v: str) -> str:
+        if v not in PROVIDER_DEFAULTS:
+            raise ValueError(
+                f"Unknown provider_id {v!r}. Must be one of: {list(PROVIDER_DEFAULTS.keys())}"
+            )
+        return v
+
+
+class TestConnectionRequest(BaseModel):
+    """Request body for POST /api/admin/ai-config/test-connection.
+
+    Unsaved form values (api_key, base_url, model_name) override the DB row so
+    admins can verify credentials before saving.  All override fields are optional;
+    omitting them falls back to whatever is stored in system_settings.
+    """
+
+    model_config = ConfigDict(extra="forbid")
+
+    provider_id: str
+    api_key: Optional[str] = None      # If non-empty, used instead of stored api_key_enc
+    base_url: Optional[str] = None     # If non-None, overrides DB base_url
+    model_name: Optional[str] = None   # If non-empty, overrides DB model_name
+
+    @field_validator("provider_id")
+    @classmethod
+    def provider_must_be_known(cls, v: str) -> str:
+        if v not in PROVIDER_DEFAULTS:
+            raise ValueError(
+                f"Unknown provider_id {v!r}. Must be one of: {list(PROVIDER_DEFAULTS.keys())}"
+            )
+        return v
+
+
+class SystemTopicCreate(BaseModel):
+    """Request model for admin system topic creation (D-09)."""
+
+    name: str
+    description: str = ""
+    color: str = "#6366f1"
+
+
+class UserDeleteConfirm(BaseModel):
+    """Admin password confirmation required before hard-deleting a user (ADMIN-02, T-05-11-01)."""
+
+    admin_password: str = Field(..., min_length=1)
+
+
+# ── SEC-08: Safe CloudConnection response model ───────────────────────────────
+
+class CloudConnectionOut(BaseModel):
+    """SEC-08: credentials_enc deliberately excluded from this response model.
+
+    Any admin or user endpoint returning CloudConnection ORM objects MUST use
+    this model to prevent accidental exposure of encrypted credentials.
+    Safe-by-default: whitelist of allowed fields (not blacklist).
+
+    Note: id is declared as str and coerced via validator so UUID ORM values
+    serialize correctly without json_encoders (Rule 1 fix — T-05-06 test suite).
+    """
+
+    id: str
+    provider: str
+    display_name: str
+    status: str
+    connected_at: datetime
+    server_url: Optional[str] = None
+    connection_username: Optional[str] = None
+    model_config = {"from_attributes": True}
+
+    @field_validator("id", mode="before")
+    @classmethod
+    def coerce_id_to_str(cls, v) -> str:
+        """Coerce UUID objects to str so the model validates from ORM instances."""
+        return str(v)
+
+
+# ── Endpoints ─────────────────────────────────────────────────────────────────
+
+
+@router.get("/users")
+async def list_users(
+    session: AsyncSession = Depends(get_db),
+    _admin: User = Depends(get_current_admin),
+) -> dict:
+    """List all users, ordered by created_at DESC.
+
+    Response shape: { items: [...safe user fields...] }
+    Never includes password_hash, credentials_enc, or document content (T-02-27).
+    """
+    result = await session.execute(
+        select(User).order_by(User.created_at.desc())
+    )
+    users = result.scalars().all()
+    return {"items": [_user_to_dict(u) for u in users]}
+
+
+@router.post("/users", status_code=status.HTTP_201_CREATED)
+async def create_user(
+    request: Request,
+    body: UserCreate,
+    session: AsyncSession = Depends(get_db),
+    _admin: User = Depends(get_current_admin),
+) -> dict:
+    """Admin creates a new user account (ADMIN-01).
+
+    - password_must_change=True forces the user to change their password on
+      first login (T-02-32, D-06).
+    - Quota row initialized at 100 MB (D-06).
+    - Returns 409 if email or handle is already taken.
+    """
+    # Check uniqueness
+    existing_email = await session.execute(
+        select(User).where(User.email == str(body.email))
+    )
+    if existing_email.scalar_one_or_none() is not None:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="Email already registered",
+        )
+
+    existing_handle = await session.execute(
+        select(User).where(User.handle == body.handle)
+    )
+    if existing_handle.scalar_one_or_none() is not None:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="Handle already taken",
+        )
+
+    new_user = User(
+        id=uuid.uuid4(),
+        handle=body.handle,
+        email=str(body.email),
+        password_hash=hash_password(body.password),
+        role=body.role,
+        is_active=True,
+        totp_enabled=False,
+        password_must_change=True,  # ADMIN-01: force password change on first login
+    )
+    session.add(new_user)
+
+    quota = Quota(
+        user_id=new_user.id,
+        limit_bytes=_DEFAULT_QUOTA_BYTES,
+        used_bytes=0,
+    )
+    session.add(quota)
+    await session.flush()  # persist User + Quota before audit_log FK references them
+    # D-13: admin user created event
+    _ip_addr = get_client_ip(request)
+    await write_audit_log(
+        session,
+        event_type="admin.user_created",
+        user_id=new_user.id,
+        actor_id=_admin.id,
+        resource_id=new_user.id,
+        ip_address=_ip_addr,
+    )
+    await session.commit()
+
+    return {
+        "id": str(new_user.id),
+        "handle": new_user.handle,
+        "email": new_user.email,
+        "role": new_user.role,
+        "created_at": new_user.created_at.isoformat() if new_user.created_at else None,
+    }
+
+
+@router.patch("/users/{user_id}/status")
+async def update_user_status(
+    user_id: uuid.UUID,
+    body: UserStatusUpdate,
+    request: Request,
+    session: AsyncSession = Depends(get_db),
+    _admin: User = Depends(get_current_admin),
+) -> dict:
+    """Deactivate or reactivate a user account (ADMIN-02).
+
+    - Prevents deactivating the last active admin (T-02-29).
+    - On deactivation: all refresh tokens are revoked (family revocation).
+    """
+    user = await session.get(User, user_id)
+    if user is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
+
+    # Guard: cannot deactivate the only remaining active admin (T-02-29)
+    if not body.is_active and user.role == "admin":
+        count_result = await session.execute(
+            select(func.count(User.id)).where(
+                User.role == "admin",
+                User.is_active.is_(True),
+            )
+        )
+        active_admin_count = count_result.scalar_one()
+        if active_admin_count <= 1:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Cannot deactivate the only admin",
+            )
+
+    _ip_addr = get_client_ip(request)
+    user.is_active = body.is_active
+
+    if not body.is_active:
+        # Revoke all refresh tokens on deactivation
+        await revoke_all_refresh_tokens(session, user.id)
+        # Revoke any pre-deactivation access tokens still within their TTL (T-7.2-01)
+        await request.app.state.redis.set(
+            f"user_nbf:{user.id}",
+            int(time.time()),
+            ex=settings.access_token_expire_minutes * 60,
+        )
+
+    session.add(user)
+
+    # D-13: user deactivated/activated event
+    _event = "admin.user_deactivated" if not body.is_active else "admin.user_activated"
+    await write_audit_log(
+        session,
+        event_type=_event,
+        user_id=user.id,
+        actor_id=_admin.id,
+        resource_id=user.id,
+        ip_address=_ip_addr,
+    )
+    await session.commit()
+
+    return {
+        "id": str(user.id),
+        "handle": user.handle,
+        "email": user.email,
+        "is_active": user.is_active,
+    }
+
+
+@router.post("/users/{user_id}/password-reset", status_code=status.HTTP_202_ACCEPTED)
+async def initiate_password_reset(
+    user_id: uuid.UUID,
+    session: AsyncSession = Depends(get_db),
+    _admin: User = Depends(get_current_admin),
+) -> dict:
+    """Admin initiates a password reset for a user (ADMIN-03).
+
+    Sends the reset email via Celery. Does NOT:
+      - return a reset token (T-02-30)
+      - grant admin access to the account
+      - log in as the target user (ADMIN-07 — no impersonation)
+
+    Returns 202 immediately regardless of email delivery status.
+    """
+    user = await session.get(User, user_id)
+    if user is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
+
+    from services.auth import create_password_reset_token  # noqa: PLC0415
+    from config import settings as _settings  # noqa: PLC0415
+
+    reset_token = create_password_reset_token(str(user.id))
+    reset_link = f"{_settings.frontend_url}/password-reset/confirm?token={reset_token}"
+
+    # Deferred import to avoid circular imports (same pattern as document_tasks)
+    from tasks.email_tasks import send_reset_email  # noqa: PLC0415
+    send_reset_email.delay(user.email, reset_link)
+
+    return {"message": "Password reset email sent"}
+
+
+@router.get("/users/{user_id}/quota")
+async def get_user_quota(
+    user_id: uuid.UUID,
+    session: AsyncSession = Depends(get_db),
+    _admin: User = Depends(get_current_admin),
+) -> dict:
+    """Return quota details for a user (ADMIN-04).
+
+    Quota info is admin-visible operational data — no PII, no document content
+    (T-02-31 disposition: accept).
+    """
+    quota = await session.get(Quota, user_id)
+    if quota is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Quota not found")
+
+    return {
+        "user_id": str(quota.user_id),
+        "limit_bytes": quota.limit_bytes,
+        "used_bytes": quota.used_bytes,
+        "limit_mb": quota.limit_bytes // 1048576,
+        "used_mb": quota.used_bytes // 1048576,
+    }
+
+
+@router.patch("/users/{user_id}/quota")
+async def update_user_quota(
+    user_id: uuid.UUID,
+    body: QuotaUpdate,
+    request: Request,
+    session: AsyncSession = Depends(get_db),
+    _admin: User = Depends(get_current_admin),
+) -> dict:
+    """Adjust a user's storage quota (ADMIN-04).
+
+    If the new limit is below current usage, still applies the change but
+    returns warning=True with an explanatory message. Uploads will be blocked
+    but existing documents are preserved.
+    """
+    quota = await session.get(Quota, user_id)
+    if quota is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Quota not found")
+
+    warning = body.limit_bytes < quota.used_bytes
+    warning_message = (
+        "New limit is below current usage. Uploads will be blocked but existing documents are preserved."
+        if warning
+        else None
+    )
+
+    _ip_addr = get_client_ip(request)
+    old_limit = quota.limit_bytes
+    quota.limit_bytes = body.limit_bytes
+    session.add(quota)
+
+    # D-13: quota changed event
+    await write_audit_log(
+        session,
+        event_type="admin.quota_changed",
+        user_id=user_id,
+        actor_id=_admin.id,
+        resource_id=None,
+        ip_address=_ip_addr,
+        metadata_={"old_bytes": old_limit, "new_bytes": body.limit_bytes},
+    )
+    await session.commit()
+
+    response: dict = {
+        "user_id": str(quota.user_id),
+        "limit_bytes": quota.limit_bytes,
+        "used_bytes": quota.used_bytes,
+        "warning": warning,
+    }
+    if warning_message:
+        response["message"] = warning_message
+    return response
+
+
+@router.patch("/users/{user_id}/ai-config")
+async def update_ai_config(
+    user_id: uuid.UUID,
+    body: UserAiConfigUpdate,
+    request: Request,
+    session: AsyncSession = Depends(get_db),
+    _admin: User = Depends(get_current_admin),
+) -> dict:
+    """Assign AI provider and model for a user (ADMIN-05).
+
+    Users cannot change their own AI provider or model (PROJECT.md Key Decision).
+    Only admins have this capability.
+    """
+    user = await session.get(User, user_id)
+    if user is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
+
+    _ip_addr = get_client_ip(request)
+    user.ai_provider = body.ai_provider
+    user.ai_model = body.ai_model
+    session.add(user)
+
+    # D-13: AI provider assigned event
+    await write_audit_log(
+        session,
+        event_type="admin.ai_provider_assigned",
+        user_id=user_id,
+        actor_id=_admin.id,
+        resource_id=None,
+        ip_address=_ip_addr,
+        metadata_={"provider": body.ai_provider, "model": body.ai_model},
+    )
+    await session.commit()
+
+    return {
+        "id": str(user.id),
+        "email": user.email,
+        "ai_provider": user.ai_provider,
+        "ai_model": user.ai_model,
+    }
+
+
+@router.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
+async def delete_user(
+    user_id: uuid.UUID,
+    body: UserDeleteConfirm,
+    request: Request,
+    session: AsyncSession = Depends(get_db),
+    _admin: User = Depends(get_current_admin),
+) -> None:
+    """Delete a user account and clean up all their MinIO objects (SEC-09, D-19).
+
+    Security invariants:
+      - Admin password verified via Argon2 before any deletion (T-05-11-01)
+      - Cannot delete admin accounts (T-04-07-04)
+      - MinIO objects are deleted BEFORE DB records are removed (SEC-09)
+      - MinIO deletion is best-effort (try/except) — DB row is deleted regardless
+      - Audit log written with event_type="admin.user_deleted"
+    """
+    # T-05-11-01: Verify admin password before performing any destructive action.
+    # Fail fast — no DB reads for the target user until the admin is confirmed.
+    if not verify_password(body.admin_password, _admin.password_hash):
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Invalid admin password",
+        )
+
+    user = await session.get(User, user_id)
+    if user is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
+
+    # T-04-07-04: Cannot delete admin accounts
+    if user.role == "admin":
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Cannot delete admin accounts",
+        )
+
+    _ip_addr = get_client_ip(request)
+
+    # SEC-09 (cloud): purge cloud-stored documents and credentials BEFORE DB delete.
+    # Must run before MinIO cleanup so that credentials are still available to build
+    # the cloud backend instances for delete_object calls.
+    cloud_conns_result = await session.execute(
+        select(CloudConnection).where(CloudConnection.user_id == user_id)
+    )
+    cloud_conns = cloud_conns_result.scalars().all()
+    for conn in cloud_conns:
+        # Delete cloud objects stored in this provider for this user
+        cloud_docs_result = await session.execute(
+            select(Document).where(
+                Document.user_id == user_id,
+                Document.storage_backend == conn.provider,
+            )
+        )
+        for doc in cloud_docs_result.scalars().all():
+            try:
+                backend = await get_storage_backend_for_document(doc, user, session)
+                await backend.delete_object(doc.object_key)
+            except Exception:
+                pass  # Best-effort cloud object cleanup; deletion proceeds regardless
+        # Purge the credentials row (FK cascade would also remove it, but explicit
+        # deletion here guarantees credentials_enc is gone before commit — SEC-09)
+        await session.delete(conn)
+    if cloud_conns:
+        await session.flush()  # Flush connection deletes before user delete
+        await write_audit_log(
+            session,
+            event_type="cloud.credentials_purged",
+            user_id=user_id,
+            actor_id=_admin.id,
+            resource_id=user_id,
+            ip_address=_ip_addr,
+            metadata_={"providers": [c.provider for c in cloud_conns]},
+        )
+
+    # SEC-09 (minio): collect all user documents and delete MinIO objects BEFORE DB delete
+    docs_result = await session.execute(
+        select(Document).where(Document.user_id == user_id)
+    )
+    user_docs = docs_result.scalars().all()
+
+    storage = get_storage_backend()
+    for doc in user_docs:
+        try:
+            await storage.delete_object(doc.object_key)
+        except Exception:
+            pass  # Best-effort MinIO cleanup; DB deletion proceeds regardless
+
+    # D-13: audit log BEFORE deleting the user row (user FK still valid at flush time)
+    await write_audit_log(
+        session,
+        event_type="admin.user_deleted",
+        user_id=user_id,
+        actor_id=_admin.id,
+        resource_id=user_id,
+        ip_address=_ip_addr,
+    )
+    await session.flush()
+
+    # Delete user record (CASCADE removes quota, documents, refresh_tokens, etc.)
+    await session.delete(user)
+    await session.commit()
+
+
+@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:
+    """Create a system topic visible to all users (D-09, DOC-04).
+
+    System topics have user_id = NULL, making them visible to every user as
+    defaults in their topic namespace. Only admins can create system topics.
+    Regular users create per-user topics via POST /api/topics.
+
+    Deduplication: case-insensitive match within the system namespace (user_id IS NULL).
+    Returns the existing system topic if one with the same name already exists.
+    """
+    from services import storage  # noqa: PLC0415
+
+    topic = await storage.create_topic(
+        session, body.name, body.description, body.color, user_id=None
+    )
+    return topic
+
+
+# ── System AI Provider Configuration (D-08, D-15) ────────────────────────────
+
+@router.get("/ai-config/models")
+async def get_ai_config_models(
+    provider_id: str,
+    session: AsyncSession = Depends(get_db),
+    _admin: User = Depends(get_current_admin),
+) -> dict:
+    """Return the list of model IDs available from a provider's API (D-08).
+
+    Calls the provider's standard GET /models endpoint using the stored
+    config (base_url + api_key from system_settings).  Always returns 200
+    with {"models": [...]} — never 5xx on provider failure (returns empty list).
+
+    Security: requires get_current_admin; provider_id from query param only;
+    decrypted api_key never appears in the response.
+    """
+    import httpx  # noqa: PLC0415 — local import keeps admin.py startup fast
+
+    config = await load_provider_config_by_id(session, provider_id)
+
+    # Resolve base_url: prefer DB row, fall back to PROVIDER_DEFAULTS
+    if config and config.base_url:
+        base_url = config.base_url.rstrip("/")
+    else:
+        base_url = (PROVIDER_DEFAULTS.get(provider_id, {}).get("base_url") or "").rstrip("/")
+
+    if not base_url:
+        return {"models": [], "provider_id": provider_id}
+
+    api_key = config.api_key if config else ""
+
+    # Build request headers — Anthropic uses x-api-key; all others use Bearer
+    if provider_id == "anthropic":
+        headers = {
+            "x-api-key": api_key,
+            "anthropic-version": "2023-06-01",
+        }
+        models_url = "https://api.anthropic.com/v1/models"
+    else:
+        headers = {"Authorization": f"Bearer {api_key}"} if api_key else {}
+        models_url = f"{base_url}/models"
+
+    try:
+        async with httpx.AsyncClient(timeout=8.0) as client:
+            resp = await client.get(models_url, headers=headers)
+            resp.raise_for_status()
+            data = resp.json()
+
+        # Standard OpenAI-compat shape: {"data": [{"id": "...", ...}, ...]}
+        # Anthropic shape: {"data": [{"id": "...", ...}, ...]}
+        # Ollama OpenAI-compat: same shape
+        raw_list = data.get("data") or data.get("models") or []
+        model_ids: list[str] = sorted(
+            {
+                item["id"] if isinstance(item, dict) else str(item)
+                for item in raw_list
+                if item
+            }
+        )
+        return {"models": model_ids, "provider_id": provider_id}
+    except Exception as exc:
+        return {"models": [], "provider_id": provider_id, "error": str(exc)[:120]}
+
+
+@router.post("/ai-config/test-connection")
+async def test_ai_connection(
+    body: TestConnectionRequest,
+    session: AsyncSession = Depends(get_db),
+    _admin: User = Depends(get_current_admin),
+) -> dict:
+    """Test connectivity for an AI provider, optionally with unsaved form values (D-08).
+
+    Loads the stored system_settings row for body.provider_id, then overlays any
+    non-empty values from the request body so admins can verify credentials before
+    saving them to the database.
+
+    Override priority (highest → lowest):
+      1. body.api_key / base_url / model_name  (unsaved form values)
+      2. system_settings DB row               (previously saved config)
+      3. PROVIDER_DEFAULTS                    (built-in fallback)
+
+    Returns {"ok": true/false, "provider_id": str} — never raises 5xx for
+    provider-side failures; surfaces as ok=False so the UI shows a clear status.
+
+    Security: requires get_current_admin; api_key from body is used only for the
+    in-flight health_check() call and is never stored or logged.
+    """
+    provider_id = body.provider_id
+    stored = await load_provider_config_by_id(session, provider_id)
+    defaults = PROVIDER_DEFAULTS.get(provider_id, {})
+
+    # Resolve effective values: body overrides DB, DB overrides PROVIDER_DEFAULTS
+    effective_api_key = (
+        body.api_key
+        if body.api_key
+        else (stored.api_key if stored else "")
+    )
+    effective_base_url = (
+        body.base_url
+        if body.base_url is not None
+        else (stored.base_url if stored else defaults.get("base_url"))
+    )
+    effective_model = (
+        body.model_name
+        if body.model_name
+        else (stored.model if stored else defaults.get("model", ""))
+    )
+
+    effective_config = ProviderConfig(
+        provider_id=provider_id,
+        api_key=effective_api_key,
+        base_url=effective_base_url,
+        model=effective_model,
+    )
+
+    try:
+        provider = get_provider(effective_config)
+        ok = await provider.health_check()
+        return {"ok": ok, "provider_id": provider_id}
+    except Exception:
+        return {"ok": False, "provider_id": provider_id, "reason": "health_check_failed"}
+
+
+@router.get("/ai-config")
+async def get_ai_config(
+    session: AsyncSession = Depends(get_db),
+    _admin: User = Depends(get_current_admin),
+) -> dict:
+    """Return all AI provider configurations for the admin panel (D-08).
+
+    Includes DB rows for providers that have been saved AND synthesised stubs
+    for providers that only exist in PROVIDER_DEFAULTS (so the admin UI always
+    shows all 10 providers even before any have been configured).
+
+    Security invariant: api_key_enc is NEVER returned (T-07-01).
+    Use has_api_key (bool) as the only indicator that a key is stored.
+    """
+    result = await session.execute(select(SystemSettings))
+    db_rows = result.scalars().all()
+
+    # Build a lookup for DB rows
+    db_by_provider: dict[str, SystemSettings] = {r.provider_id: r for r in db_rows}
+
+    providers_out = []
+    for pid in PROVIDER_DEFAULTS:
+        if pid in db_by_provider:
+            providers_out.append(_ai_config_to_dict(db_by_provider[pid]))
+        else:
+            # Synthesise a stub entry for providers with no DB row yet
+            defaults = PROVIDER_DEFAULTS[pid]
+            providers_out.append({
+                "provider_id": pid,
+                "base_url": defaults.get("base_url"),
+                "model_name": defaults.get("model", ""),
+                "context_chars": defaults.get("context_chars", 8000),
+                "is_active": False,
+                "has_api_key": False,
+                "updated_at": None,
+            })
+
+    return {"providers": providers_out}
+
+
+@router.put("/ai-config")
+async def update_system_ai_config(
+    body: SystemAiConfigUpdate,
+    request: Request,
+    session: AsyncSession = Depends(get_db),
+    _admin: User = Depends(get_current_admin),
+) -> dict:
+    """Create or update a system-level AI provider configuration (D-08, D-15).
+
+    Upsert semantics: if no row exists for body.provider_id, one is created using
+    PROVIDER_DEFAULTS for any omitted fields.
+
+    API key handling (T-07-01 mitigated):
+      - body.api_key is None → leave existing api_key_enc untouched
+      - body.api_key == "" → clear api_key_enc (set to NULL)
+      - body.api_key is a non-empty string → HKDF-encrypt and store
+
+    is_active=True handling (T-07-03 mitigated):
+      When body.is_active is True, a single atomic UPDATE flips all rows:
+      SET is_active = (provider_id = :target_id)
+      This guarantees COUNT(WHERE is_active) == 1 with no read-then-write race.
+
+    Audit log (T-07-14 mitigated):
+      metadata_ contains only provider_id + fields_changed list — never the
+      api_key value itself.
+    """
+    from config import settings as _settings  # noqa: PLC0415
+
+    # Load existing row or create a new one from PROVIDER_DEFAULTS
+    stmt = select(SystemSettings).where(SystemSettings.provider_id == body.provider_id)
+    result = await session.execute(stmt)
+    row = result.scalar_one_or_none()
+
+    is_new = row is None
+    if is_new:
+        defaults = PROVIDER_DEFAULTS[body.provider_id]
+        row = SystemSettings(
+            provider_id=body.provider_id,
+            model_name=defaults.get("model", ""),
+            context_chars=defaults.get("context_chars", 8000),
+            base_url=defaults.get("base_url"),
+            is_active=False,
+            api_key_enc=None,
+        )
+
+    # Track which fields the caller explicitly set (for audit log — never api_key value)
+    fields_changed: list[str] = []
+
+    # Apply provided fields
+    if body.api_key is not None:
+        fields_changed.append("api_key")
+        if body.api_key == "":
+            row.api_key_enc = None
+        else:
+            master_key_str = _settings.cloud_creds_key
+            master_key_bytes = (
+                master_key_str.encode("utf-8")
+                if isinstance(master_key_str, str)
+                else master_key_str
+            )
+            row.api_key_enc = encrypt_api_key(master_key_bytes, body.provider_id, body.api_key)
+
+    if body.base_url is not None:
+        row.base_url = body.base_url
+        fields_changed.append("base_url")
+
+    if body.model_name is not None:
+        row.model_name = body.model_name
+        fields_changed.append("model_name")
+
+    if body.context_chars is not None:
+        row.context_chars = body.context_chars
+        fields_changed.append("context_chars")
+
+    if body.is_active is not None:
+        fields_changed.append("is_active")
+
+    if is_new:
+        session.add(row)
+        await session.flush()  # ensure row has an id before UPDATE
+
+    # Atomic is_active flip: SET is_active = (provider_id = :target) on ALL rows.
+    # Single UPDATE statement prevents dual-active race condition (T-07-03).
+    if body.is_active is True:
+        await session.execute(
+            update(SystemSettings).values(
+                is_active=(SystemSettings.provider_id == body.provider_id)
+            )
+        )
+        # Reflect the flip on the in-memory row
+        row.is_active = True
+
+    _ip_addr = get_client_ip(request)
+    await write_audit_log(
+        session,
+        event_type="admin.ai_config_changed",
+        user_id=None,
+        actor_id=_admin.id,
+        resource_id=None,
+        ip_address=_ip_addr,
+        metadata_={"provider_id": body.provider_id, "fields_changed": fields_changed},
+    )
+
+    await session.commit()
+
+    # Reload to pick up DB-generated updated_at after commit
+    await session.refresh(row)
+
+    return _ai_config_to_dict(row)