"""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, Literal

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)
    # Inherit ``remember_me`` from the old refresh token's ``rmb`` claim
    # so a 30-day session stays 30-day across rotations.  Old tokens
    # issued before this fix lack the claim — default to False (7-day)
    # which is the previous behaviour, so no regression for existing
    # sessions.
    inherited_remember_me = bool(refresh_payload.get("rmb", False))
    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=inherited_remember_me,
        )
        await svc.rotate(
            old_refresh_token=payload.refresh_token,
            new_refresh_token=new_pair.refresh_token,
            new_ttl_seconds=_refresh_ttl_seconds(inherited_remember_me),
        )
    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}


# ---------------------------------------------------------------------------
# SSO routes — OIDC redirect/callback + SAML ACS (U4)
# ---------------------------------------------------------------------------


class SSORedirectResponse(BaseModel):
    """SSO 重定向响应 — 返回 IDP authorize URL 供前端跳转。"""

    model_config = ConfigDict(extra="forbid")

    redirect_url: str
    state: str


async def _sso_issue_token_pair(request: Request, user_dict: dict[str, object]) -> TokenResponse:
    """SSO 登录成功后签发 JWT pair + 创建 session（复用 login 流程）。"""
    # SEC-1: SSO 流程必须与本地 login（auth.py:377）一致地校验 is_active，
    # 否则已 deprovision 的用户可通过 IDP 回调重新拿到 JWT + session。
    if not bool(user_dict.get("is_active", True)):
        raise HTTPException(status_code=403, detail="Account is disabled")
    db_path = await _ensure_db(request)
    secret = _resolve_jwt_secret(request)
    svc: SessionService = get_session_service()
    info = _client_info(request)

    pre_session = await svc.create(
        SessionCreate(
            user_id=str(user_dict["id"]),
            refresh_token="__pending_sso__",
            device_fingerprint=info["fingerprint"],
            device_label=info["label"],
            ip=info["ip"],
            user_agent=info["user_agent"],
            auth_provider="sso",
            ttl_seconds=_refresh_ttl_seconds(False),
        )
    )
    token_pair = create_token_pair(
        user_id=str(user_dict["id"]),
        username=str(user_dict["username"]),
        role=str(user_dict["role"]),
        secret=secret,
        session_id=pre_session.id,
    )
    async with aiosqlite.connect(str(db_path)) as db:
        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(), str(user_dict["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=str(user_dict["id"]),
            username=str(user_dict["username"]),
            email=str(user_dict["email"]),
            role=str(user_dict["role"]),
            is_active=bool(user_dict.get("is_active", True)),
            is_terminal_authorized=bool(user_dict.get("is_terminal_authorized", False)),
            is_server_terminal_authorized=bool(
                user_dict.get("is_server_terminal_authorized", False)
            ),
        ),
        session_id=pre_session.id,
    )


@router.get("/oauth/{provider}/redirect", response_model=SSORedirectResponse)
async def oidc_redirect(provider: str, request: Request) -> SSORedirectResponse:
    """OIDC 重定向 — 返回 IDP authorize URL（前端跳转）。"""
    from agentkit.server.auth.providers.oidc import OIDCProvider

    try:
        url = await OIDCProvider().get_redirect_url(provider)
    except ValueError as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    # state 已由 provider 写入 _state_cache，从 URL 中提取回传前端
    from urllib.parse import parse_qs, urlparse

    parsed = urlparse(url)
    state_values = parse_qs(parsed.query).get("state", [])
    state = state_values[0] if state_values else ""
    return SSORedirectResponse(redirect_url=url, state=state)


@router.get("/oauth/{provider}/callback", response_model=TokenResponse)
async def oidc_callback(
    provider: str,
    code: str,
    state: str,
    request: Request,
) -> TokenResponse:
    """OIDC callback — code 换 token + userinfo + JIT 用户关联 → JWT pair。"""
    await _ensure_db(request)
    from agentkit.server.auth.providers.oidc import OIDCProvider

    try:
        user_dict = await OIDCProvider().handle_callback(provider, code, state)
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    return await _sso_issue_token_pair(request, user_dict)


class SAMLACSRequest(BaseModel):
    """SAML ACS 请求体。"""

    model_config = ConfigDict(extra="forbid")

    SAMLResponse: str
    RelayState: str | None = None


@router.post("/saml/{provider}/acs", response_model=TokenResponse)
async def saml_acs(
    provider: str,
    payload: SAMLACSRequest,
    request: Request,
) -> TokenResponse:
    """SAML ACS — 消费 SAMLResponse + NameID 提取 + JIT 用户关联 → JWT pair。"""
    await _ensure_db(request)
    from agentkit.server.auth.providers.saml import SAMLProvider

    request_url = str(request.url)
    try:
        user_dict = await SAMLProvider().handle_acs(
            provider, payload.SAMLResponse, request_url=request_url
        )
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    return await _sso_issue_token_pair(request, user_dict)


# ---------------------------------------------------------------------------
# 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}


# ---------------------------------------------------------------------------
# Admin deprovisioning + RBAC role change (U4 — R1b / KTD-8)
# ---------------------------------------------------------------------------


class DeprovisionRequest(BaseModel):
    """手动 deprovisioning 请求体。"""

    model_config = ConfigDict(extra="forbid")

    reason: str | None = None
    break_glass: bool = False  # 高权限账户 break-glass 标记


@admin_router.post("/users/{user_id}/deprovision")
async def deprovision_user(
    user_id: str,
    payload: DeprovisionRequest,
    request: Request,
    admin: dict[str, Any] = Depends(_require_admin),
) -> dict[str, Any]:
    """手动 deprovisioning (R1b) — operator/admin RBAC 强制。

    1. 禁用本地用户 (is_active=0)
    2. 撤销该用户所有活跃 session
    3. 记录审计 (actor/target/reason/source/timestamp)，接入 OTel
    4. break_glass=True 时记录 source=break_glass（高权限账户应急通道）
    """
    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")
        await db.execute(
            "UPDATE users SET is_active = 0, updated_at = ? WHERE id = ?",
            (_now_iso(), user_id),
        )
        await db.commit()
    # 撤销所有 session
    svc: SessionService = get_session_service()
    revoked = await svc.revoke_all_for_user(user_id, reason="admin_revoked")
    # 审计记录
    from agentkit.server.auth.audit import (
        SOURCE_BREAKGLASS,
        SOURCE_MANUAL,
        get_audit_service,
    )

    source = SOURCE_BREAKGLASS if payload.break_glass else SOURCE_MANUAL
    await get_audit_service().record_deprovision(
        actor_user_id=str(admin.get("user_id") or ""),
        actor_username=str(admin.get("username") or ""),
        target_user_id=user_id,
        target_username=str(row["username"]),
        reason=payload.reason,
        source=source,
    )
    return {"deprovisioned": True, "revoked_sessions": revoked}


class RoleChangeRequest(BaseModel):
    """RBAC 角色变更请求体。"""

    model_config = ConfigDict(extra="forbid")

    role: Literal["member", "operator", "admin"]
    reason: str | None = None


@admin_router.post("/users/{user_id}/role")
async def change_user_role(
    user_id: str,
    payload: RoleChangeRequest,
    request: Request,
    admin: dict[str, Any] = Depends(_require_admin),
) -> dict[str, Any]:
    """RBAC 角色变更 (KTD-8) — admin only。

    记录 actor/target/role_before/role_after/reason/source/timestamp，接入 OTel 审计。
    变更后撤销该用户所有现有 session（强制重新登录以刷新 JWT role claim）。
    """
    db_path = await _ensure_db(request)
    from agentkit.server.auth.audit import SOURCE_MANUAL, get_audit_service

    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")
        role_before = str(row["role"])
        await db.execute(
            "UPDATE users SET role = ?, updated_at = ? WHERE id = ?",
            (payload.role, _now_iso(), user_id),
        )
        await db.commit()
    # 审计记录
    await get_audit_service().record_role_change(
        actor_user_id=str(admin.get("user_id") or ""),
        actor_username=str(admin.get("username") or ""),
        target_user_id=user_id,
        target_username=str(row["username"]),
        role_before=role_before,
        role_after=payload.role,
        reason=payload.reason,
        source=SOURCE_MANUAL,
    )
    # 撤销所有 session（强制重新登录刷新 role claim）
    svc: SessionService = get_session_service()
    revoked = await svc.revoke_all_for_user(user_id, reason="role_changed")
    return {
        "role_changed": True,
        "role_before": role_before,
        "role_after": payload.role,
        "revoked_sessions": revoked,
    }


# 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)