"""Authentication REST routes (U4 — Centralized Auth & Token Persistence).

Endpoints
---------
- ``POST /auth/login``        — username + password (+ remember_me) → JWT pair + session
- ``POST /auth/refresh``      — refresh token → new access token (rotated session)
- ``POST /auth/logout``       — revoke current session
- ``GET  /auth/whoami``       — return the current user (used for cold-start)
- ``GET  /auth/sessions``     — list the current user's active sessions
- ``DELETE /auth/sessions/{id}`` — revoke a specific session (own only)
- ``POST /auth/change-password`` — change password + revoke all sessions
- ``GET  /admin/sessions``    — list all sessions (admin)
- ``DELETE /admin/sessions/{id}`` — revoke a session as admin
- ``GET  /admin/users/{user_id}/sessions`` — list a specific user's sessions (admin)
- ``DELETE /admin/users/{user_id}/sessions/{session_id}`` — revoke a specific user's session (admin)

The auth DB (SQLite via aiosqlite) and JWT secret are resolved from
``app.state`` if set by the app factory, otherwise from the defaults in
:mod:`agentkit.server.auth.models` / :mod:`agentkit.server.auth.jwt_utils`.
"""

from __future__ import annotations

import hmac
import logging
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

import aiosqlite
import jwt
from fastapi import APIRouter, Depends, HTTPException, Request
from pydantic import BaseModel, ConfigDict, EmailStr

from agentkit.server.auth.dependencies import require_authenticated
from agentkit.server.auth.jwt_utils import (
    ACCESS_TOKEN_TTL,
    REFRESH_TOKEN_TTL,
    REFRESH_TOKEN_TTL_REMEMBER_ME,
    create_access_token,
    create_token_pair,
    get_or_create_jwt_secret,
    verify_token,
)
from agentkit.server.auth.models import (
    auth_session_row_to_dict,
    DEFAULT_AUTH_DB_PATH,
    init_auth_db,
)
from agentkit.server.auth.password import hash_password, verify_password
from agentkit.server.auth.permissions import Permission, has_permission
from agentkit.server.auth.session_service import (
    REVOKE_REASON_PASSWORD_CHANGED,
    REVOKE_REASON_USER_TERMINATED,
    SessionCreate,
    SessionNotFound,
    SessionReuseDetected,
    SessionService,
    get_session_service,
)
from agentkit.server.auth.denylist import hash_token

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/auth", tags=["auth"])

# Paths under /auth that the AuthMiddleware whitelists. The middleware
# already knows about /auth/login, /auth/refresh, /auth/logout; the new
# /auth/whoami requires a valid token so it's NOT whitelisted.
_AUTH_PUBLIC_PATHS = ("/auth/login", "/auth/refresh")


# ---------------------------------------------------------------------------
# Pydantic request/response models
# ---------------------------------------------------------------------------


class LoginRequest(BaseModel):
    """Username + password login payload."""

    model_config = ConfigDict(extra="forbid")

    username: str
    password: str
    remember_me: bool = False


class RefreshRequest(BaseModel):
    """Refresh-token exchange payload."""

    model_config = ConfigDict(extra="forbid")

    refresh_token: str


class LogoutRequest(BaseModel):
    """Logout payload — accepts the current refresh token to revoke."""

    model_config = ConfigDict(extra="forbid")

    refresh_token: str


class ChangePasswordRequest(BaseModel):
    """Change-password payload — old + new password."""

    model_config = ConfigDict(extra="forbid")

    old_password: str
    new_password: str


class UserResponse(BaseModel):
    """Public user representation (no password hash)."""

    model_config = ConfigDict(extra="forbid")

    id: str
    username: str
    email: EmailStr
    role: str
    is_active: bool
    is_terminal_authorized: bool
    is_server_terminal_authorized: bool


class SessionResponse(BaseModel):
    """Public projection of an ``auth_sessions`` row."""

    model_config = ConfigDict(extra="forbid")

    id: str
    device_fingerprint: str
    device_label: str
    ip: str
    user_agent: str
    auth_provider: str
    created_at: str
    last_active_at: str
    expires_at: str
    is_current: bool = False
    revoked: bool = False
    revoked_reason: str | None = None
    user_id: str | None = None
    previous_session_id: str | None = None


class TokenResponse(BaseModel):
    """JWT pair + user info returned by /auth/login and /auth/refresh."""

    model_config = ConfigDict(extra="forbid")

    access_token: str
    refresh_token: str
    token_type: str = "bearer"
    expires_in: int = int(ACCESS_TOKEN_TTL.total_seconds())
    user: UserResponse
    session_id: str


class WhoamiResponse(BaseModel):
    """Response of ``GET /auth/whoami`` (cold-start + session metadata).

    When the client calls whoami with a refresh token (cold-start), the
    server issues a fresh access token so the client doesn't need a
    separate ``/auth/refresh`` round-trip.  When called with an access
    token, ``access_token`` is ``None`` (the caller already has one).
    """

    model_config = ConfigDict(extra="forbid")

    user: UserResponse
    access_token: str | None = None
    session_id: str | None = None
    session: SessionResponse | None = None


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


def _resolve_db_path(request: Request) -> Path:
    path = getattr(request.app.state, "auth_db_path", None)
    return Path(path) if path else DEFAULT_AUTH_DB_PATH


def _resolve_jwt_secret(request: Request) -> str:
    secret = getattr(request.app.state, "jwt_secret", None)
    if secret:
        return secret
    return get_or_create_jwt_secret()


async def _ensure_db(request: Request) -> Path:
    """Resolve the auth DB path and ensure its schema is initialized.

    Always calls ``init_auth_db`` (idempotent via ``CREATE TABLE IF NOT
    EXISTS``) so partial / pre-migration databases get missing tables
    added on the next request (U5 — R8).  Setting ``PRAGMA busy_timeout``
    inside ``init_auth_db`` also protects against concurrent writes.
    """
    db_path = _resolve_db_path(request)
    await init_auth_db(db_path)
    return db_path


def _user_row_to_response(row: aiosqlite.Row) -> UserResponse:
    return UserResponse(
        id=row["id"],
        username=row["username"],
        email=row["email"],
        role=row["role"],
        is_active=bool(row["is_active"]),
        is_terminal_authorized=bool(row["is_terminal_authorized"]),
        is_server_terminal_authorized=bool(row["is_server_terminal_authorized"]),
    )


def _refresh_ttl_seconds(remember_me: bool) -> int:
    return (
        int(REFRESH_TOKEN_TTL_REMEMBER_ME.total_seconds())
        if remember_me
        else int(REFRESH_TOKEN_TTL.total_seconds())
    )


# U10 rollout: clients whose X-Client-Version header reports a version
# below this cutoff receive legacy (no-sid) tokens so that they
# continue to work during the migration window. After the cutoff is
# bumped, the legacy code path is removed (Phase 5 cleanup).
LEGACY_CLIENT_VERSION_CUTOFF = "0.5.0"


def _is_legacy_client(request: Request) -> bool:
    """Return True if the request's client version is below the cutoff.

    The check is intentionally lenient: missing header, unparseable
    version, or any error evaluating the version all default to
    ``False`` (treat as a new client). The only path that returns
    ``True`` is a syntactically valid semver that compares strictly
    less than the cutoff. This avoids accidentally downgrading
    legitimate clients with unusual but new-enough versions.
    """
    raw = request.headers.get("X-Client-Version", "").strip()
    if not raw:
        return False
    try:
        client_v = _parse_semver(raw)
        cutoff_v = _parse_semver(LEGACY_CLIENT_VERSION_CUTOFF)
        if client_v is None or cutoff_v is None:
            return False
        return client_v < cutoff_v
    except (ValueError, TypeError, AttributeError):  # noqa: BLE001
        logger.debug("Failed to parse X-Client-Version %r", raw)
        return False


def _parse_semver(value: str) -> tuple[int, int, int] | None:
    """Parse ``"MAJOR.MINOR.PATCH"`` into a comparable tuple.

    Returns ``None`` when the value is not a strict 3-part semver.
    Pre-release / build-metadata are ignored — we only need a coarse
    ordering for the rollout policy.
    """
    parts = value.split(".")
    if len(parts) != 3:
        return None
    try:
        return (int(parts[0]), int(parts[1]), int(parts[2]))
    except ValueError:
        return None


# ---------------------------------------------------------------------------
# Client-info extraction (device fingerprint / IP / user agent)
# ---------------------------------------------------------------------------


def _client_info(request: Request) -> dict[str, str]:
    """Extract device fingerprint, IP and user-agent from the request.

    The fingerprint is a coarse best-effort hash of the User-Agent +
    Accept-Language — enough to distinguish "Mac/Safari" from
    "Windows/Chrome" in the sessions list without storing anything
    sensitive.  Production deployments that need stronger device
    identification can layer that on top later.
    """
    ua = request.headers.get("user-agent", "")
    al = request.headers.get("accept-language", "")
    fp_seed = f"{ua}|{al}"
    # Use the same sha256 helper as the denylist for consistency.
    fp = hash_token(fp_seed)[:16]
    # Best-effort client IP (no X-Forwarded-For trust for now).
    ip = (request.client.host if request.client else "") or ""
    label = _device_label(ua)
    return {
        "fingerprint": fp,
        "label": label,
        "ip": ip,
        "user_agent": ua[:512],
    }


def _device_label(user_agent: str) -> str:
    """Human-readable device label, e.g. ``"macOS — Chrome 124"``."""
    ua = user_agent.lower()
    if "edg/" in ua:
        browser = "Edge"
    elif "chrome" in ua and "chromium" in ua:
        browser = "Chrome"
    elif "firefox" in ua:
        browser = "Firefox"
    elif "safari" in ua and "chrome" not in ua:
        browser = "Safari"
    elif "tauri" in ua or "fischer" in ua:
        return "Fischer Desktop"
    else:
        browser = "Browser"
    if "mac os x" in ua or "macintosh" in ua:
        os_ = "macOS"
    elif "windows" in ua:
        os_ = "Windows"
    elif "iphone" in ua or "ipad" in ua:
        os_ = "iOS"
    elif "android" in ua:
        os_ = "Android"
    elif "linux" in ua:
        os_ = "Linux"
    else:
        os_ = "Unknown"
    return f"{os_} — {browser}"


# ---------------------------------------------------------------------------
# Routes
# ---------------------------------------------------------------------------


@router.post("/login", response_model=TokenResponse)
async def login(payload: LoginRequest, request: Request) -> TokenResponse:
    """Authenticate with username + password and receive a JWT pair.

    Flow:
    1. Validate via the configured :class:`AuthProvider`.
    2. Issue access + refresh JWTs (with ``sid`` / ``jti``).
    3. Persist the session to ``auth_sessions`` (V2).
    4. Update ``last_login_at``.

    U10 back-compat: when the caller's ``X-Client-Version`` header is
    below the rollout cutoff, the issued tokens carry no ``sid`` claim
    and the session row is removed.  The client can still authenticate
    using the legacy ``user_sessions`` table; new clients always get
    the V2 flow.  See ``docs/migrations/2026-06-20-client-version-rollout.md``.
    """
    db_path = await _ensure_db(request)
    secret = _resolve_jwt_secret(request)
    from agentkit.server.auth.providers import (
        InvalidCredentials,
        LocalAuthProvider,
    )

    # 1. Authenticate (construct a provider bound to this request's
    # auth DB so the route works for both the global singleton and
    # per-request overrides such as test fixtures that swap the path).
    try:
        user = await LocalAuthProvider(db_path=db_path).authenticate(
            username=payload.username, password=payload.password
        )
    except InvalidCredentials as exc:
        raise HTTPException(status_code=401, detail=str(exc)) from exc

    if not user.is_active:
        raise HTTPException(status_code=403, detail="Account is disabled")

    # 2. Issue tokens
    # The session_id is created up front (SessionService.create returns
    # the id) so we can sign the tokens with the same id used in the
    # session row.
    svc: SessionService = get_session_service()
    info = _client_info(request)

    # U10 rollout policy: detect old clients by the X-Client-Version
    # header and downgrade them to the legacy (no-sid) token shape.
    # Old clients can't handle the new flow (they don't call
    # /auth/whoami with a refresh token), so handing them a sid-bearing
    # token would just look invalid to them.
    legacy_mode = _is_legacy_client(request)

    # Pre-create the session with a placeholder row using the id we are
    # about to issue, so the JWT's ``sid`` claim is consistent with the
    # row's primary key.  We use SessionService directly because the
    # refresh_token has not been issued yet — we let create() generate
    # the row first, then issue the tokens referencing it.
    #
    # Actually: simpler — call create() AFTER issuing the tokens, then
    # re-issue nothing; just persist the row with the known id.

    # Step A: create the session row with a placeholder hash.  The real
    # hash will be written in a follow-up UPDATE; this is a tiny race
    # window that no one can exploit because the row has no valid
    # refresh_token_hash to match.
    pre_session = await svc.create(
        SessionCreate(
            user_id=user.id,
            refresh_token="__pending__",  # overwritten in Step C
            device_fingerprint=info["fingerprint"],
            device_label=info["label"],
            ip=info["ip"],
            user_agent=info["user_agent"],
            auth_provider="local",
            ttl_seconds=_refresh_ttl_seconds(payload.remember_me),
        )
    )

    # Step B: issue tokens using the session id from Step A.
    token_pair = create_token_pair(
        user_id=user.id,
        username=user.username,
        role=user.role,
        secret=secret,
        session_id=pre_session.id,
        remember_me=payload.remember_me,
        legacy_mode=legacy_mode,
    )

    # Step C: overwrite the row with the real refresh token hash.
    import aiosqlite as _aio

    async with _aio.connect(str(db_path)) as db:
        if legacy_mode:
            # U10: a legacy client cannot use a session row (no
            # X-Client-Version-aware client logic), so drop the V2
            # session we just created.  Legacy refresh-tokens will be
            # tracked via the existing user_sessions table on the next
            # refresh (the /auth/refresh route handles both shapes).
            await db.execute(
                "DELETE FROM auth_sessions WHERE id = ?",
                (pre_session.id,),
            )
        else:
            await db.execute(
                "UPDATE auth_sessions SET refresh_token_hash = ? WHERE id = ?",
                (hash_token(token_pair.refresh_token), pre_session.id),
            )
        await db.execute(
            "UPDATE users SET last_login_at = ?, updated_at = ? WHERE id = ?",
            (_now_iso(), _now_iso(), user.id),
        )
        await db.commit()

    return TokenResponse(
        access_token=token_pair.access_token,
        refresh_token=token_pair.refresh_token,
        expires_in=int(ACCESS_TOKEN_TTL.total_seconds()),
        user=UserResponse(
            id=user.id,
            username=user.username,
            email=user.email,
            role=user.role,
            is_active=user.is_active,
            is_terminal_authorized=user.is_terminal_authorized,
            is_server_terminal_authorized=user.is_server_terminal_authorized,
        ),
        session_id="" if legacy_mode else pre_session.id,
    )


@router.post("/refresh", response_model=TokenResponse)
async def refresh(payload: RefreshRequest, request: Request) -> TokenResponse:
    """Exchange a valid refresh token for a new access token.

    Implements refresh-token rotation:
    1. Verify the refresh token's JWT signature and ``type`` claim.
    2. Look up the session by the token's SHA-256 hash.
    3. Verify the session is not revoked or expired.
    4. Issue a new access + refresh pair.
    5. Replace the session's stored hash with the new token's hash.
    6. Add the old hash to the denylist (reuse detection).

    On reuse (old hash in the denylist) the caller's
    :class:`SessionService.rotate` revokes all of that user's sessions
    and raises :class:`SessionReuseDetected`.  The route maps that to
    a 401 so the client is forced to re-authenticate.
    """
    db_path = await _ensure_db(request)
    secret = _resolve_jwt_secret(request)
    svc: SessionService = get_session_service()

    # 1. Verify signature + type
    try:
        refresh_payload = verify_token(payload.refresh_token, secret, expected_type="refresh")
    except (jwt.PyJWTError, ValueError, KeyError) as exc:  # noqa: BLE001
        raise HTTPException(status_code=401, detail="Invalid refresh token") from exc

    # 2-3. Validate the session (also handles reuse detection)
    try:
        new_pair = create_token_pair(
            user_id=refresh_payload["sub"],
            username=refresh_payload["username"],
            role=refresh_payload["role"],
            secret=secret,
            session_id=refresh_payload.get("sid"),
            remember_me=False,
        )
        await svc.rotate(
            old_refresh_token=payload.refresh_token,
            new_refresh_token=new_pair.refresh_token,
            new_ttl_seconds=int(REFRESH_TOKEN_TTL.total_seconds()),
        )
    except (SessionReuseDetected, SessionNotFound, ValueError, KeyError, RuntimeError) as exc:  # noqa: BLE001 — SessionReuseDetected / SessionNotFound
        logger.info("Refresh rejected: %s", exc)
        raise HTTPException(status_code=401, detail="Invalid refresh token") from exc

    # Re-fetch the user to surface fresh role / is_active
    async with aiosqlite.connect(str(db_path)) as db:
        db.row_factory = aiosqlite.Row
        cursor = await db.execute("SELECT * FROM users WHERE id = ?", (refresh_payload["sub"],))
        user_row = await cursor.fetchone()
    if user_row is None or not bool(user_row["is_active"]):
        raise HTTPException(status_code=401, detail="User not found or disabled")

    return TokenResponse(
        access_token=new_pair.access_token,
        refresh_token=new_pair.refresh_token,
        expires_in=int(ACCESS_TOKEN_TTL.total_seconds()),
        user=_user_row_to_response(user_row),
        session_id=refresh_payload.get("sid", ""),
    )


@router.post("/logout")
async def logout(payload: LogoutRequest, request: Request) -> dict[str, Any]:
    """Revoke the refresh-token session. Idempotent."""
    svc: SessionService = get_session_service()
    revoked = await svc.revoke_by_refresh_token(
        payload.refresh_token, reason=REVOKE_REASON_USER_TERMINATED
    )
    return {"revoked": revoked}


@router.get("/whoami", response_model=WhoamiResponse)
async def whoami(request: Request) -> WhoamiResponse:
    """Return the current user + session metadata (cold-start support).

    Accepts **either** an access token (normal call) **or** a refresh
    token (cold-start, when the access token is gone).  The route does
    its own auth because the middleware only accepts access tokens.

    On cold-start (refresh token presented), the server issues a fresh
    access token so the client doesn't need a separate ``/auth/refresh``
    round-trip.  On 401 from this endpoint, the client treats it as
    'invalid' state (NOT 'error' state) so the router redirects to
    /login.
    """
    auth_header = request.headers.get("Authorization", "")
    if not auth_header.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="missing bearer token")
    token = auth_header[7:]

    secret = _resolve_jwt_secret(request)
    try:
        # Accept both access and refresh tokens for cold-start.
        payload = verify_token(token, secret, expected_type=None)
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="token expired")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="invalid token")

    user_id = payload.get("sub")
    if not user_id:
        raise HTTPException(status_code=401, detail="invalid token: no subject")

    token_type = payload.get("type")
    sid = payload.get("sid")

    # Load the user row.
    db_path = await _ensure_db(request)
    async with aiosqlite.connect(str(db_path)) as db:
        db.row_factory = aiosqlite.Row
        cursor = await db.execute("SELECT * FROM users WHERE id = ?", (user_id,))
        row = await cursor.fetchone()
    if row is None or not bool(row["is_active"]):
        raise HTTPException(status_code=401, detail="User not found or disabled")
    user_response = _user_row_to_response(row)

    # V2 token with sid: validate the session is still active.
    session_response: SessionResponse | None = None
    if sid:
        svc: SessionService = get_session_service()
        info = await svc.get(sid)
        if info is None or info.revoked:
            raise HTTPException(status_code=401, detail="session revoked or expired")
        # Cold-start defense (R9): when the presented token is a refresh
        # token, verify its hash matches the session's current
        # ``refresh_token_hash``.  If the token has been rotated (via
        # ``/auth/refresh``) or revoked, the stored hash will differ and
        # we reject the request.  Comparison uses ``hmac.compare_digest``
        # for constant-time equality to prevent timing attacks.
        if token_type == "refresh":
            stored_hash = await svc.get_stored_refresh_hash(sid)
            if stored_hash is None:
                raise HTTPException(status_code=401, detail="session not found")
            presented_hash = hash_token(token)
            if not hmac.compare_digest(presented_hash, stored_hash):
                raise HTTPException(
                    status_code=401,
                    detail="refresh token has been rotated or revoked",
                )
        session_response = SessionResponse(
            **auth_session_row_to_dict(_info_to_dict(info)),
            is_current=True,
        )

    # Cold-start: refresh token presented → issue a fresh access token.
    # Use ``create_access_token`` (not ``create_token_pair``) to avoid
    # minting a new refresh token that would silently be discarded —
    # the client already holds a valid refresh token.
    new_access_token: str | None = None
    if token_type == "refresh":
        new_access_token = create_access_token(
            user_id=str(row["id"]),
            username=str(row["username"]),
            role=str(row["role"]),
            secret=secret,
            session_id=sid,
        )

    return WhoamiResponse(
        user=user_response,
        access_token=new_access_token,
        session_id=sid,
        session=session_response,
    )


@router.get("/sessions", response_model=list[SessionResponse])
async def list_sessions(
    request: Request,
    user: dict[str, Any] = Depends(require_authenticated),
) -> list[SessionResponse]:
    """List the current user's active sessions."""
    user_id = user.get("user_id")
    if not user_id:
        raise HTTPException(status_code=401, detail="Authentication required")
    svc: SessionService = get_session_service()
    current_sid = user.get("sid")  # the calling session's id, if V2
    sessions = await svc.list_for_user(user_id)
    return [
        SessionResponse(
            **auth_session_row_to_dict(
                # build a dict-like object with the right keys
                _info_to_dict(s),
            ),
            is_current=(s.id == current_sid),
        )
        for s in sessions
    ]


def _info_to_dict(s: Any) -> dict[str, Any]:
    """Convert a SessionInfo to the dict shape :func:`auth_session_row_to_dict` expects."""
    return {
        "id": s.id,
        "user_id": s.user_id,
        "device_fingerprint": s.device_fingerprint,
        "device_label": s.device_label,
        "ip": s.ip,
        "user_agent": s.user_agent,
        "auth_provider": s.auth_provider,
        "created_at": s.created_at,
        "last_active_at": s.last_active_at,
        "expires_at": s.expires_at,
        "revoked": s.revoked,
        "revoked_reason": s.revoked_reason,
        "previous_session_id": s.previous_session_id,
    }


@router.delete("/sessions/{session_id}")
async def revoke_own_session(
    session_id: str,
    request: Request,
    user: dict[str, Any] = Depends(require_authenticated),
) -> dict[str, Any]:
    """Revoke one of the current user's sessions (e.g. a lost device)."""
    user_id = user.get("user_id")
    if not user_id:
        raise HTTPException(status_code=401, detail="Authentication required")
    svc: SessionService = get_session_service()
    info = await svc.get(session_id)
    if info is None or info.user_id != user_id:
        # Don't reveal whether the session exists for another user.
        raise HTTPException(status_code=404, detail="Session not found")
    ok = await svc.revoke(session_id, reason=REVOKE_REASON_USER_TERMINATED)
    return {"revoked": ok}


@router.post("/logout-others")
async def logout_others(
    request: Request,
    user: dict[str, Any] = Depends(require_authenticated),
) -> dict[str, Any]:
    """Revoke all of the current user's sessions except the calling one.

    Used by the "Log out other devices" button in the security settings.
    The current session (identified by the JWT's ``sid`` claim) is spared.
    """
    user_id = user.get("user_id")
    if not user_id:
        raise HTTPException(status_code=401, detail="Authentication required")
    current_sid = user.get("sid")
    if not current_sid:
        # Legacy token without sid — can't identify the current session.
        raise HTTPException(
            status_code=400,
            detail="Current session cannot be identified (legacy token). Please log in again.",
        )
    svc: SessionService = get_session_service()
    count = await svc.revoke_all_for_user(
        user_id,
        except_sid=current_sid,
        reason=REVOKE_REASON_USER_TERMINATED,
    )
    return {"revoked_count": count}


@router.post("/change-password")
async def change_password(
    payload: ChangePasswordRequest,
    request: Request,
    user: dict[str, Any] = Depends(require_authenticated),
) -> dict[str, Any]:
    """Change the current user's password.

    Per the security policy: changing the password invalidates ALL of
    the user's other active sessions (the user is re-authenticated on
    this device on next request via the new password).
    """
    user_id = user.get("user_id")
    if not user_id:
        raise HTTPException(status_code=401, detail="Authentication required")
    if len(payload.new_password) < 8:
        raise HTTPException(status_code=400, detail="new_password must be at least 8 characters")
    db_path = await _ensure_db(request)
    async with aiosqlite.connect(str(db_path)) as db:
        db.row_factory = aiosqlite.Row
        cursor = await db.execute("SELECT * FROM users WHERE id = ?", (user_id,))
        row = await cursor.fetchone()
        if row is None:
            raise HTTPException(status_code=404, detail="User not found")
        if not verify_password(payload.old_password, row["password_hash"]):
            raise HTTPException(status_code=400, detail="Old password is incorrect")
        new_hash = hash_password(payload.new_password)
        await db.execute(
            "UPDATE users SET password_hash = ?, updated_at = ? WHERE id = ?",
            (new_hash, _now_iso(), user_id),
        )
        await db.commit()
    # Revoke all sessions except the current one (the user is staying
    # logged in on the device that just changed the password).
    svc: SessionService = get_session_service()
    current_sid = user.get("sid")
    sessions = await svc.list_for_user(user_id)
    revoked = 0
    for s in sessions:
        if s.id != current_sid:
            if await svc.revoke(s.id, reason=REVOKE_REASON_PASSWORD_CHANGED):
                revoked += 1
    return {"changed": True, "revoked_other_sessions": revoked}


# ---------------------------------------------------------------------------
# Admin routes
# ---------------------------------------------------------------------------

admin_router = APIRouter(prefix="/admin", tags=["admin"])


async def _require_admin(
    request: Request, user: dict[str, Any] = Depends(require_authenticated)
) -> dict[str, Any]:
    if not has_permission(user, Permission.USER_MANAGE):
        raise HTTPException(status_code=403, detail="Admin permission required")
    return user


@admin_router.get("/sessions", response_model=list[SessionResponse])
async def admin_list_sessions(
    request: Request,
    limit: int = 200,
    admin: dict[str, Any] = Depends(_require_admin),
) -> list[SessionResponse]:
    """List recent sessions across all users (admin only)."""
    svc: SessionService = get_session_service()
    sessions = await svc.list_all(include_revoked=False, limit=limit)
    return [
        SessionResponse(
            **auth_session_row_to_dict(_info_to_dict(s)),
            is_current=False,
        )
        for s in sessions
    ]


@admin_router.delete("/sessions/{session_id}")
async def admin_revoke_session(
    session_id: str,
    request: Request,
    admin: dict[str, Any] = Depends(_require_admin),
) -> dict[str, Any]:
    """Revoke any session as admin (force-logout a user)."""
    svc: SessionService = get_session_service()
    ok = await svc.revoke(session_id, reason="admin_revoked")
    if not ok:
        raise HTTPException(status_code=404, detail="Session not found or already revoked")
    return {"revoked": True}


@admin_router.get(
    "/users/{user_id}/sessions",
    response_model=list[SessionResponse],
)
async def admin_list_user_sessions(
    user_id: str,
    request: Request,
    include_revoked: bool = False,
    admin: dict[str, Any] = Depends(_require_admin),
) -> list[SessionResponse]:
    """List all sessions for a specific user (admin only).

    By default only active sessions are returned. When ``include_revoked``
    is true, revoked sessions are also returned (with their ``revoked``
    and ``revoked_reason`` fields populated in the response).

    The endpoint verifies the user exists; if not, 404 is returned. This
    prevents admins from silently filtering on a stale id and gives
    clearer feedback.
    """
    db_path = await _ensure_db(request)
    async with aiosqlite.connect(str(db_path)) as db:
        db.row_factory = aiosqlite.Row
        cursor = await db.execute("SELECT id, username FROM users WHERE id = ?", (user_id,))
        user_row = await cursor.fetchone()
    if user_row is None:
        raise HTTPException(status_code=404, detail="User not found")

    svc: SessionService = get_session_service()
    sessions = await svc.list_for_user(user_id, include_revoked=include_revoked)
    current_sid = admin.get("sid")
    return [
        SessionResponse(
            **auth_session_row_to_dict(_info_to_dict(s)),
            is_current=(s.id == current_sid),
        )
        for s in sessions
    ]


@admin_router.delete("/users/{user_id}/sessions/{session_id}")
async def admin_revoke_user_session(
    user_id: str,
    session_id: str,
    request: Request,
    admin: dict[str, Any] = Depends(_require_admin),
) -> dict[str, Any]:
    """Revoke a specific session for a specific user (admin only).

    Returns 404 if the session does not exist or does not belong to
    ``user_id`` (defense against accidentally revoking a session on the
    wrong user).  Already-revoked sessions return 200 with
    ``revoked=False`` so the admin can detect no-op vs new revoke.
    """
    svc: SessionService = get_session_service()
    info = await svc.get(session_id)
    if info is None or info.user_id != user_id:
        raise HTTPException(status_code=404, detail="Session not found for user")
    ok = await svc.revoke(session_id, reason="admin_revoked")
    return {"revoked": ok}


# Backwards-compat /me endpoint (kept for the existing frontend)
@router.get("/me", response_model=UserResponse)
async def me(
    request: Request,
    user: dict[str, Any] = Depends(require_authenticated),
) -> UserResponse:
    """Alias for :func:`whoami` (legacy path, access-token only).

    Returns just the user profile (no session metadata) for backwards
    compatibility with clients that expect the old ``UserResponse``
    shape from ``/auth/me``.
    """
    user_id = user.get("user_id")
    if not user_id:
        raise HTTPException(status_code=401, detail="Authentication required")
    db_path = await _ensure_db(request)
    async with aiosqlite.connect(str(db_path)) as db:
        db.row_factory = aiosqlite.Row
        cursor = await db.execute("SELECT * FROM users WHERE id = ?", (user_id,))
        row = await cursor.fetchone()
    if row is None:
        raise HTTPException(status_code=404, detail="User not found")
    return _user_row_to_response(row)
