vignette/apps/api/app/auth_sessions.py

"""Server-side browser sessions for the BFF auth boundary.

Sessions are DB-backed when PostgreSQL is available and fall back to the
process-local store for local degraded development. The browser only receives an
opaque HttpOnly cookie; roles and user identity stay on the server.
"""

from __future__ import annotations

import hashlib
import hmac
import logging
import secrets
import time
import uuid
from datetime import datetime, timezone
from dataclasses import dataclass, replace

from .auth_types import AccountStatus, RoleName
from .config import settings
from .db import get_pool
from .runtime_policy import require_runtime_fallback_allowed, runtime_fallback_allowed
from .runtime_schema import runtime_schema_bootstrap_required
from .services import notifications

logger = logging.getLogger(__name__)


@dataclass(slots=True)
class SessionUser:
    user_id: str
    email: str
    display_name: str
    role: RoleName
    admin_access: bool
    super_admin: bool
    account_status: AccountStatus
    cohort_ids: list[str]
    consent_at: float | None
    profile_completed_at: float | None
    expires_at: float
    learner_feedback_enabled: bool = True


@dataclass(slots=True)
class ManagedUser:
    user_id: str
    email: str
    display_name: str
    role: RoleName
    admin_access: bool
    account_status: AccountStatus
    cohort_ids: list[str]
    affiliation: str
    legal_name: str
    department: str
    grade_level: str
    phone: str
    contact_address: str
    nickname: str
    self_introduction: str
    avatar_url: str
    consent_at: float | None
    profile_completed_at: float | None
    terms_agreed_at: float | None
    privacy_agreed_at: float | None
    terms_version: str
    privacy_version: str
    created_at: float
    last_seen_at: float
    learner_feedback_enabled: bool = True


@dataclass(slots=True)
class ManagedUserMemoryInput:
    email: str
    display_name: str
    role: RoleName
    admin_access: bool | None = None
    account_status: AccountStatus | None = None
    cohort_ids: list[str] | None = None
    user_id: str | None = None
    affiliation: str | None = None
    legal_name: str | None = None
    department: str | None = None
    grade_level: str | None = None
    phone: str | None = None
    contact_address: str | None = None
    nickname: str | None = None
    self_introduction: str | None = None
    avatar_url: str | None = None
    consent_at: float | None = None
    profile_completed_at: float | None = None
    terms_agreed_at: float | None = None
    privacy_agreed_at: float | None = None
    terms_version: str | None = None
    privacy_version: str | None = None
    learner_feedback_enabled: bool | None = None
    reactivate: bool = False

    @classmethod
    def from_user(
        cls,
        user: ManagedUser,
        *,
        reactivate: bool = False,
    ) -> "ManagedUserMemoryInput":
        return cls(
            email=user.email,
            display_name=user.display_name,
            role=user.role,
            admin_access=user.admin_access,
            account_status=user.account_status,
            cohort_ids=list(user.cohort_ids),
            user_id=user.user_id,
            affiliation=user.affiliation,
            legal_name=user.legal_name,
            department=user.department,
            grade_level=user.grade_level,
            phone=user.phone,
            contact_address=user.contact_address,
            nickname=user.nickname,
            self_introduction=user.self_introduction,
            avatar_url=user.avatar_url,
            consent_at=user.consent_at,
            profile_completed_at=user.profile_completed_at,
            terms_agreed_at=user.terms_agreed_at,
            privacy_agreed_at=user.privacy_agreed_at,
            terms_version=user.terms_version,
            privacy_version=user.privacy_version,
            learner_feedback_enabled=user.learner_feedback_enabled,
            reactivate=reactivate,
        )


@dataclass(slots=True)
class ManagedUserUpsertInput:
    email: str
    display_name: str
    role: RoleName
    admin_access: bool | None = None
    cohort_ids: list[str] | None = None
    user_id: str | None = None
    external_id: str | None = None
    affiliation: str | None = None
    account_status: AccountStatus | None = None
    learner_feedback_enabled: bool | None = None
    reactivate: bool = False


@dataclass(slots=True)
class ManagedUserPatch:
    display_name: str | None = None
    role: RoleName | None = None
    admin_access: bool | None = None
    account_status: AccountStatus | None = None
    affiliation: str | None = None
    legal_name: str | None = None
    department: str | None = None
    grade_level: str | None = None
    phone: str | None = None
    contact_address: str | None = None
    nickname: str | None = None
    self_introduction: str | None = None
    avatar_url: str | None = None
    cohort_ids: list[str] | None = None
    complete_onboarding: bool = False
    terms_version: str | None = None
    privacy_version: str | None = None
    learner_feedback_enabled: bool | None = None


class InactiveUserError(Exception):
    """Raised when an inactive managed user attempts to create a login session."""


_sessions: dict[str, SessionUser] = {}
_users: dict[str, ManagedUser] = {}
_email_index: dict[str, str] = {}
_auth_identity_alias_index: dict[str, str] = {}
_inactive_emails: set[str] = set()

DEFAULT_AFFILIATION = settings.default_affiliation.strip()
DB_ROLE_BY_APP = {"learner": "learner", "teacher": "instructor", "admin": "admin"}
APP_ROLE_BY_DB = {"learner": "learner", "instructor": "teacher", "admin": "admin"}
VALID_ACCOUNT_STATUSES: set[str] = {"pending", "approved", "suspended"}


def _sid_hash(raw_sid: str) -> str:
    return hmac.new(
        settings.session_secret.encode("utf-8"),
        raw_sid.encode("utf-8"),
        hashlib.sha256,
    ).hexdigest()


def user_id_from_email(email: str) -> str:
    return str(uuid.uuid5(uuid.NAMESPACE_URL, f"vignette:user:{email.strip().lower()}"))


def user_id_from_external_id(external_id: str) -> str:
    return str(
        uuid.uuid5(
            uuid.NAMESPACE_URL, f"vignette:user-external:{external_id.strip().lower()}"
        )
    )


def _normalize_email(email: str) -> str:
    return email.strip().lower()


def _email_domain(email: str) -> str:
    normalized = _normalize_email(email)
    if "@" not in normalized:
        return ""
    return normalized.rsplit("@", 1)[1]


def _allowed_email_domain_set() -> set[str]:
    return {
        normalized
        for value in settings.auth_allowed_email_domains
        if (normalized := str(value).strip().lower().lstrip("@"))
    }


def _normalize_external_id(external_id: str | None, email: str) -> str:
    value = (external_id or "").strip().lower()
    return value or f"email:{_normalize_email(email)}"


def _db_role(role: str) -> str:
    return DB_ROLE_BY_APP.get(role, role)


def _app_role(role: str) -> str:
    return APP_ROLE_BY_DB.get(role, role)


def _account_status(value: str | None) -> AccountStatus:
    normalized = (value or "").strip().lower()
    if normalized in VALID_ACCOUNT_STATUSES:
        return normalized  # type: ignore[return-value]
    return "approved"


def _normalize_email_set(values: list[str]) -> set[str]:
    return {email for value in values if (email := _normalize_email(value))}


def _auto_approved_email_set() -> set[str]:
    return (
        _normalize_email_set(settings.auth_super_admin_emails)
        | _normalize_email_set(settings.auth_admin_emails)
        | _normalize_email_set(settings.auth_teacher_emails)
        | _normalize_email_set(settings.auth_approved_emails)
    )


def is_super_admin_email(email: str) -> bool:
    """설정 기반 신뢰 루트인 슈퍼 관리자 계정인지 판정한다."""
    normalized_email = _normalize_email(email)
    return normalized_email in _normalize_email_set(settings.auth_super_admin_emails)


def has_admin_access(
    email: str,
    role: str | None = None,
    stored_admin_access: bool = False,
) -> bool:
    """기본 역할을 유지한 채 관리자 화면에 들어갈 수 있는 계정인지 판정한다."""
    normalized_email = _normalize_email(email)
    return (
        stored_admin_access
        or role == "admin"
        or is_super_admin_email(normalized_email)
        or normalized_email in _normalize_email_set(settings.auth_admin_emails)
    )


def _is_dev_login_external_id(external_id: str | None) -> bool:
    return (external_id or "").strip().lower().startswith("dev:")


def _initial_account_status(
    *,
    email: str,
    external_id: str | None,
    reactivate: bool,
) -> AccountStatus:
    normalized_email = _normalize_email(email)
    if reactivate or normalized_email in _auto_approved_email_set():
        return "approved"
    # 로컬 E2E 자동 승인은 허용된 내부 도메인에만 적용한다. 관리자 exact-email로
    # 사전등록한 외부 연구참여자의 pending 상태를 dev 로그인이 승격하면 안 된다.
    if (
        settings.environment == "dev"
        and _is_dev_login_external_id(external_id)
        and _email_domain(normalized_email) in _allowed_email_domain_set()
    ):
        return "approved"
    return _account_status(settings.auth_new_user_default_status)


def _ts(value: datetime | None) -> float:
    return (value or datetime.now(timezone.utc)).timestamp()


def _optional_ts(value: datetime | None) -> float | None:
    return value.timestamp() if value is not None else None


def _row_value(row, key: str, default=None):
    try:
        return row[key]
    except (IndexError, KeyError, TypeError):
        return default


def _cohort_ids(cohort: str | None) -> list[str]:
    """단일 TEXT 컬럼(콤마 직렬화)을 cohort id 리스트로 복원한다."""
    if not cohort:
        return []
    return [part for part in (piece.strip() for piece in cohort.split(",")) if part]


def _cohort_value(cohort_ids: list[str] | None) -> str | None:
    """cohort id 리스트를 콤마 직렬화해 저장한다. 첫 항목만 남기던 유실 버그 수정."""
    cleaned = [part for part in ((item or "").strip() for item in cohort_ids or []) if part]
    if not cleaned:
        return None
    return ",".join(cleaned)


async def _runtime_tables_ready(conn) -> bool:
    row = await conn.fetchrow(
        """
        SELECT
          EXISTS (
            SELECT 1 FROM information_schema.columns
            WHERE table_schema = 'app'
              AND table_name = 'app_user'
              AND column_name IN (
                'affiliation',
                'legal_name',
                'department',
                'grade_level',
                'phone',
                'contact_address',
                'nickname',
                'self_introduction',
                'avatar_url',
                'consent_at',
                'profile_completed_at',
                'terms_agreed_at',
                'privacy_agreed_at',
                'terms_version',
                'privacy_version',
                'last_seen_at',
                'account_status',
                'admin_access',
                'learner_feedback_enabled',
                'updated_at'
              )
            GROUP BY table_schema, table_name
            HAVING count(*) = 20
          ) AS has_user_columns,
          EXISTS (
            SELECT 1 FROM information_schema.columns
            WHERE table_schema = 'app'
              AND table_name = 'persona_card'
              AND column_name = 'triggers'
          ) AS has_persona_triggers,
          to_regclass('app.persona_voice_map') IS NOT NULL AS has_persona_voice_map,
          to_regclass('app.auth_session') IS NOT NULL AS has_auth_session,
          to_regclass('app.auth_identity_alias') IS NOT NULL AS has_auth_identity_alias,
          EXISTS (
            SELECT 1 FROM information_schema.columns
            WHERE table_schema = 'app'
              AND table_name = 'auth_session'
              AND column_name = 'login_email'
          ) AS has_auth_session_login_email,
          EXISTS (
            SELECT 1 FROM pg_policies
            WHERE schemaname = 'app'
              AND tablename = 'auth_identity_alias'
              AND policyname = 'p_auth_identity_alias_select'
          ) AS has_auth_identity_alias_select_policy,
          to_regclass('app.user_preferences') IS NOT NULL AS has_preferences,
          EXISTS (
            SELECT 1 FROM information_schema.columns
            WHERE table_schema = 'app'
              AND table_name = 'admin_engine_config'
              AND column_name = 'reasoning_effort'
          ) AS has_engine_config,
          EXISTS (
            SELECT 1 FROM information_schema.columns
            WHERE table_schema = 'app'
              AND table_name = 'sessions'
              AND column_name IN (
                'runtime_case_id',
                'persona_code',
                'persona_display_name',
                'persona_difficulty',
                'prev_rapport_credit',
                'session_goals',
                'learner_feedback_enabled'
              )
            GROUP BY table_schema, table_name
            HAVING count(*) = 7
          ) AS has_session_columns,
          EXISTS (
            SELECT 1 FROM information_schema.columns
            WHERE table_schema = 'app'
              AND table_name = 'session_state'
              AND column_name = 'turns_in_stage'
          ) AS has_state_columns,
          EXISTS (
            SELECT 1 FROM information_schema.columns
            WHERE table_schema = 'app'
              AND table_name = 'turns'
              AND column_name IN (
                'audio_ref',
                'silence_ms',
                'speech_rate',
                'barge_in',
                'provider_events'
              )
            GROUP BY table_schema, table_name
            HAVING count(*) = 5
          ) AS has_turn_voice_metadata_columns,
          EXISTS (
            SELECT 1 FROM information_schema.columns
            WHERE table_schema = 'app'
              AND table_name = 'session_review_status'
              AND column_name IN (
                'worksheet_status',
                'worksheet_note',
                'worksheet_reviewed_at'
              )
            GROUP BY table_schema, table_name
            HAVING count(*) = 3
          ) AS has_session_review_worksheet_columns,
          (
            SELECT count(*) = 4
            FROM app.stage_def
            WHERE stage_code IN ('라포','탐색','개입','정리')
          ) AS has_stage_defs,
          to_regclass('app.admin_health_event') IS NOT NULL AS has_admin_health_event,
          to_regclass('app.admin_health_daily_rollup') IS NOT NULL AS has_admin_health_daily_rollup,
          EXISTS (
            SELECT 1 FROM information_schema.columns
            WHERE table_schema = 'app'
              AND table_name = 'admin_health_daily_rollup'
              AND column_name = 'last_down_at'
          ) AS has_admin_health_daily_rollup_columns,
          to_regclass('app.support_ticket') IS NOT NULL AS has_support_ticket,
          EXISTS (
            SELECT 1 FROM information_schema.columns
            WHERE table_schema = 'app'
              AND table_name = 'support_ticket'
              AND column_name IN ('fingerprint', 'parent_ticket_id')
            GROUP BY table_schema, table_name
            HAVING count(*) = 2
          ) AS has_support_ticket_duplicate_columns,
          EXISTS (
            SELECT 1 FROM pg_policies
            WHERE schemaname = 'app'
              AND tablename = 'admin_health_event'
              AND policyname IN (
                'p_admin_health_event_select',
                'p_admin_health_event_insert',
                'p_admin_health_event_delete'
              )
            GROUP BY schemaname, tablename
            HAVING count(*) = 3
          ) AS has_admin_health_event_policies,
          EXISTS (
            SELECT 1 FROM pg_policies
            WHERE schemaname = 'app'
              AND tablename = 'admin_health_daily_rollup'
              AND policyname IN (
                'p_admin_health_daily_rollup_select',
                'p_admin_health_daily_rollup_insert',
                'p_admin_health_daily_rollup_update'
              )
            GROUP BY schemaname, tablename
            HAVING count(*) = 3
          ) AS has_admin_health_daily_rollup_policies,
          EXISTS (
            SELECT 1 FROM pg_policies
            WHERE schemaname = 'app'
              AND tablename = 'support_ticket'
              AND policyname IN (
                'p_support_ticket_select',
                'p_support_ticket_insert',
                'p_support_ticket_update',
                'p_support_ticket_delete'
              )
            GROUP BY schemaname, tablename
            HAVING count(*) = 4
          ) AS has_support_ticket_policies,
          to_regclass('app.learner_prepost_measure') IS NOT NULL AS has_learner_prepost_measure,
          EXISTS (
            SELECT 1 FROM pg_policies
            WHERE schemaname = 'app'
              AND tablename = 'learner_prepost_measure'
              AND policyname IN (
                'p_learner_prepost_measure_select',
                'p_learner_prepost_measure_insert',
                'p_learner_prepost_measure_update'
              )
            GROUP BY schemaname, tablename
            HAVING count(*) = 3
          ) AS has_learner_prepost_measure_policies,
          EXISTS (
            SELECT 1 FROM pg_policies
            WHERE schemaname = 'app'
              AND tablename = 'sessions'
              AND policyname IN ('p_sessions_insert','p_sessions_update','p_sessions_delete')
            GROUP BY schemaname, tablename
            HAVING count(*) = 3
          ) AS has_session_write_policies,
          NOT EXISTS (
            SELECT 1 FROM pg_policies
            WHERE schemaname = 'app'
              AND tablename = 'sessions'
              AND policyname = 'p_sessions_modify'
          ) AS removed_old_session_policy,
          EXISTS (
            SELECT 1 FROM pg_policies
            WHERE schemaname = 'app'
              AND tablename = 'turns'
              AND policyname IN ('p_turns_insert','p_turns_update','p_turns_delete')
            GROUP BY schemaname, tablename
            HAVING count(*) = 3
          ) AS has_turn_write_policies,
          NOT EXISTS (
            SELECT 1 FROM pg_policies
            WHERE schemaname = 'app'
              AND tablename = 'turns'
              AND policyname = 'p_turns_modify'
          ) AS removed_old_turn_policy
        """
    )
    return bool(
        row
        and row["has_user_columns"]
        and row["has_persona_triggers"]
        and row["has_persona_voice_map"]
        and row["has_auth_session"]
        and row["has_auth_identity_alias"]
        and row["has_auth_session_login_email"]
        and row["has_auth_identity_alias_select_policy"]
        and row["has_preferences"]
        and row["has_engine_config"]
        and row["has_session_columns"]
        and row["has_state_columns"]
        and row["has_turn_voice_metadata_columns"]
        and row["has_session_review_worksheet_columns"]
        and row["has_stage_defs"]
        and row["has_admin_health_event"]
        and row["has_admin_health_daily_rollup"]
        and row["has_admin_health_daily_rollup_columns"]
        and row["has_support_ticket"]
        and row["has_support_ticket_duplicate_columns"]
        and row["has_admin_health_event_policies"]
        and row["has_admin_health_daily_rollup_policies"]
        and row["has_support_ticket_policies"]
        and row["has_learner_prepost_measure"]
        and row["has_learner_prepost_measure_policies"]
        and row["has_session_write_policies"]
        and row["removed_old_session_policy"]
        and row["has_turn_write_policies"]
        and row["removed_old_turn_policy"]
    )


async def ensure_runtime_tables() -> None:
    """Ensure DB-backed auth/user runtime tables exist when a pool is available."""
    pool = get_pool()
    async with pool.acquire() as conn:
        ready = await _runtime_tables_ready(conn)
        if not runtime_schema_bootstrap_required("auth/user", ready=ready):
            return
        await conn.execute(
            """
            DO $$
            BEGIN
              IF to_regclass('app.persona_card') IS NOT NULL
                 AND NOT EXISTS (
                   SELECT 1 FROM information_schema.columns
                   WHERE table_schema = 'app'
                     AND table_name = 'persona_card'
                     AND column_name = 'triggers'
                 ) THEN
                ALTER TABLE app.persona_card
                  ADD COLUMN IF NOT EXISTS triggers JSONB NOT NULL DEFAULT '{}'::jsonb;
              END IF;

              IF to_regclass('app.persona_card') IS NOT NULL
                 AND to_regclass('app.persona_voice_map') IS NULL THEN
                CREATE TABLE IF NOT EXISTS app.persona_voice_map (
                  persona_id UUID NOT NULL,
                  version INT NOT NULL,
                  voice_id TEXT NOT NULL,
                  provider TEXT NOT NULL CHECK (provider IN ('openai','higgs','melotts')),
                  base_params JSONB NOT NULL,
                  prosody_map JSONB NOT NULL,
                  PRIMARY KEY (persona_id, version),
                  FOREIGN KEY (persona_id, version)
                    REFERENCES app.persona_card(persona_id, version) ON DELETE CASCADE
                );
              END IF;

              IF EXISTS (
                SELECT 1 FROM information_schema.columns
                WHERE table_schema = 'app'
                  AND table_name = 'app_user'
                  AND column_name = 'affiliation'
                  AND column_default IS DISTINCT FROM ''''::text
              ) THEN
                ALTER TABLE app.app_user ALTER COLUMN affiliation SET DEFAULT '';
              END IF;
            END $$;
            """
        )
        if await _runtime_tables_ready(conn):
            return
        await conn.execute(
            """
            ALTER TABLE app.app_user
              ADD COLUMN IF NOT EXISTS affiliation TEXT NOT NULL DEFAULT '',
              ADD COLUMN IF NOT EXISTS legal_name TEXT NOT NULL DEFAULT '',
              ADD COLUMN IF NOT EXISTS department TEXT NOT NULL DEFAULT '',
              ADD COLUMN IF NOT EXISTS grade_level TEXT NOT NULL DEFAULT '',
              ADD COLUMN IF NOT EXISTS phone TEXT NOT NULL DEFAULT '',
              ADD COLUMN IF NOT EXISTS contact_address TEXT NOT NULL DEFAULT '',
              ADD COLUMN IF NOT EXISTS nickname TEXT NOT NULL DEFAULT '',
              ADD COLUMN IF NOT EXISTS self_introduction TEXT NOT NULL DEFAULT '',
              ADD COLUMN IF NOT EXISTS avatar_url TEXT NOT NULL DEFAULT '',
              ADD COLUMN IF NOT EXISTS consent_at TIMESTAMPTZ,
              ADD COLUMN IF NOT EXISTS profile_completed_at TIMESTAMPTZ,
              ADD COLUMN IF NOT EXISTS terms_agreed_at TIMESTAMPTZ,
              ADD COLUMN IF NOT EXISTS privacy_agreed_at TIMESTAMPTZ,
              ADD COLUMN IF NOT EXISTS terms_version TEXT NOT NULL DEFAULT '',
              ADD COLUMN IF NOT EXISTS privacy_version TEXT NOT NULL DEFAULT '',
              ADD COLUMN IF NOT EXISTS last_seen_at TIMESTAMPTZ NOT NULL DEFAULT now(),
              ADD COLUMN IF NOT EXISTS account_status TEXT NOT NULL DEFAULT 'approved',
              ADD COLUMN IF NOT EXISTS admin_access BOOLEAN NOT NULL DEFAULT false,
              ADD COLUMN IF NOT EXISTS learner_feedback_enabled BOOLEAN NOT NULL DEFAULT true,
              ADD COLUMN IF NOT EXISTS updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
            """
        )
        await conn.execute(
            """
            DO $$
            BEGIN
              IF NOT EXISTS (
                SELECT 1 FROM pg_constraint
                WHERE conname = 'app_user_account_status_check'
                  AND conrelid = 'app.app_user'::regclass
              ) THEN
                ALTER TABLE app.app_user
                  ADD CONSTRAINT app_user_account_status_check
                  CHECK (account_status IN ('pending','approved','suspended'));
              END IF;
            END $$;
            CREATE INDEX IF NOT EXISTS idx_app_user_account_status
              ON app.app_user(account_status, last_seen_at DESC);
            """
        )
        await conn.execute(
            """
            CREATE TABLE IF NOT EXISTS app.auth_session (
              sid_hash TEXT PRIMARY KEY,
              user_id UUID NOT NULL REFERENCES app.app_user(user_id) ON DELETE CASCADE,
              role TEXT NOT NULL,
              display_name TEXT NOT NULL,
              cohort_ids JSONB NOT NULL DEFAULT '[]'::jsonb,
              expires_at TIMESTAMPTZ NOT NULL,
              revoked_at TIMESTAMPTZ,
              created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
              last_seen_at TIMESTAMPTZ NOT NULL DEFAULT now()
            )
            """
        )
        await conn.execute(
            """
            ALTER TABLE app.auth_session
              ADD COLUMN IF NOT EXISTS login_email TEXT NOT NULL DEFAULT '';

            CREATE TABLE IF NOT EXISTS app.auth_identity_alias (
              external_id TEXT PRIMARY KEY
                CHECK (external_id ~ '^[a-z0-9_-]+:.+$'),
              user_id UUID NOT NULL
                REFERENCES app.app_user(user_id) ON DELETE RESTRICT,
              source_user_id UUID
                REFERENCES app.app_user(user_id) ON DELETE SET NULL,
              linked_at TIMESTAMPTZ NOT NULL DEFAULT now(),
              linked_by TEXT NOT NULL,
              reason TEXT NOT NULL,
              CHECK (source_user_id IS NULL OR source_user_id <> user_id)
            );

            CREATE INDEX IF NOT EXISTS idx_auth_identity_alias_user
              ON app.auth_identity_alias(user_id, linked_at DESC);

            ALTER TABLE app.auth_identity_alias ENABLE ROW LEVEL SECURITY;
            DROP POLICY IF EXISTS p_auth_identity_alias_select
              ON app.auth_identity_alias;
            CREATE POLICY p_auth_identity_alias_select
              ON app.auth_identity_alias
              FOR SELECT
              USING (true);
            """
        )
        await conn.execute(
            """
            CREATE INDEX IF NOT EXISTS idx_auth_session_user_active
              ON app.auth_session(user_id, expires_at)
              WHERE revoked_at IS NULL
            """
        )
        await conn.execute(
            """
            CREATE TABLE IF NOT EXISTS app.user_preferences (
              user_id UUID PRIMARY KEY REFERENCES app.app_user(user_id) ON DELETE CASCADE,
              theme TEXT NOT NULL DEFAULT 'system',
              voice_preset_id TEXT NOT NULL DEFAULT 'soft-young-fem',
              voice_rate REAL NOT NULL DEFAULT 1.0,
              notifications JSONB NOT NULL DEFAULT '{}'::jsonb,
              updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
            )
            """
        )
        await conn.execute(
            """
            CREATE TABLE IF NOT EXISTS app.admin_engine_config (
              id BOOLEAN PRIMARY KEY DEFAULT TRUE CHECK (id),
              engine_mode TEXT NOT NULL,
              engine_url TEXT NOT NULL,
              model TEXT NOT NULL,
              reasoning_effort TEXT,
              updated_by TEXT,
              updated_at TIMESTAMPTZ
            )
            """
        )
        await conn.execute(
            """
            ALTER TABLE app.admin_engine_config
              ADD COLUMN IF NOT EXISTS reasoning_effort TEXT
            """
        )
        await conn.execute(
            """
            CREATE TABLE IF NOT EXISTS app.admin_provider_credential (
              provider TEXT PRIMARY KEY,
              token_encrypted TEXT NOT NULL,
              token_hint TEXT NOT NULL DEFAULT '',
              auth_kind TEXT NOT NULL DEFAULT 'api_key'
                CHECK (auth_kind IN ('api_key','oauth_token')),
              updated_by TEXT,
              updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
              last_verified_at TIMESTAMPTZ,
              last_verify_ok BOOLEAN,
              last_verify_error TEXT
            )
            """
        )
        await conn.execute(
            """
            CREATE TABLE IF NOT EXISTS app.admin_health_event (
              id BIGSERIAL PRIMARY KEY,
              observed_at TIMESTAMPTZ NOT NULL DEFAULT now(),
              overall_status TEXT NOT NULL CHECK (overall_status IN ('ok','degraded','down')),
              environment TEXT NOT NULL,
              engine_mode TEXT NOT NULL,
              service_key TEXT NOT NULL,
              service_name TEXT NOT NULL,
              service_status TEXT NOT NULL CHECK (service_status IN ('ok','degraded','down')),
              detail TEXT NOT NULL DEFAULT '',
              metric TEXT NOT NULL DEFAULT '',
              load REAL NOT NULL DEFAULT 0.0 CHECK (load >= 0.0 AND load <= 1.0),
              captured_by UUID REFERENCES app.app_user(user_id)
            )
            """
        )
        await conn.execute(
            """
            CREATE INDEX IF NOT EXISTS idx_admin_health_event_observed
              ON app.admin_health_event(observed_at DESC, service_key)
            """
        )
        await conn.execute(
            """
            CREATE TABLE IF NOT EXISTS app.admin_health_daily_rollup (
              rollup_date DATE NOT NULL,
              environment TEXT NOT NULL,
              engine_mode TEXT NOT NULL,
              service_key TEXT NOT NULL,
              service_name TEXT NOT NULL,
              sample_count INTEGER NOT NULL DEFAULT 0 CHECK (sample_count >= 0),
              ok_samples INTEGER NOT NULL DEFAULT 0 CHECK (ok_samples >= 0),
              degraded_samples INTEGER NOT NULL DEFAULT 0 CHECK (degraded_samples >= 0),
              down_samples INTEGER NOT NULL DEFAULT 0 CHECK (down_samples >= 0),
              first_observed_at TIMESTAMPTZ NOT NULL,
              last_observed_at TIMESTAMPTZ NOT NULL,
              latest_status TEXT NOT NULL CHECK (latest_status IN ('ok','degraded','down')),
              last_down_at TIMESTAMPTZ,
              max_load REAL NOT NULL DEFAULT 0.0 CHECK (max_load >= 0.0 AND max_load <= 1.0),
              generated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
              PRIMARY KEY (rollup_date, environment, engine_mode, service_key)
            )
            """
        )
        await conn.execute(
            """
            CREATE INDEX IF NOT EXISTS idx_admin_health_daily_rollup_latest
              ON app.admin_health_daily_rollup(last_observed_at DESC, service_key)
            """
        )
        await conn.execute(
            """
            CREATE TABLE IF NOT EXISTS app.support_ticket (
              id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
              reporter_id UUID REFERENCES app.app_user(user_id) ON DELETE SET NULL,
              reporter_email TEXT NOT NULL,
              reporter_name TEXT NOT NULL,
              reporter_role TEXT NOT NULL,
              category TEXT NOT NULL DEFAULT 'other' CHECK (
                category IN (
                  'account_access',
                  'session_review',
                  'voice_browser',
                  'content_scenario',
                  'safety',
                  'other'
                )
              ),
              priority TEXT NOT NULL DEFAULT 'normal' CHECK (priority IN ('low','normal','high','urgent')),
              status TEXT NOT NULL DEFAULT 'open' CHECK (
                status IN ('open','triaged','in_progress','resolved','closed')
              ),
              subject TEXT NOT NULL,
              body TEXT NOT NULL,
              source_path TEXT NOT NULL DEFAULT '',
              fingerprint TEXT NOT NULL DEFAULT '',
              parent_ticket_id UUID REFERENCES app.support_ticket(id) ON DELETE SET NULL,
              assigned_group TEXT NOT NULL DEFAULT '',
              resolution_note TEXT NOT NULL DEFAULT '',
              created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
              updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
              resolved_at TIMESTAMPTZ,
              last_activity_at TIMESTAMPTZ NOT NULL DEFAULT now()
            )
            """
        )
        await conn.execute(
            """
            CREATE INDEX IF NOT EXISTS idx_support_ticket_status_priority
              ON app.support_ticket(status, priority, updated_at DESC)
            """
        )
        await conn.execute(
            """
            CREATE INDEX IF NOT EXISTS idx_support_ticket_reporter
              ON app.support_ticket(reporter_id, created_at DESC)
            """
        )
        await conn.execute(
            """
            ALTER TABLE app.support_ticket
              ADD COLUMN IF NOT EXISTS fingerprint TEXT NOT NULL DEFAULT '',
              ADD COLUMN IF NOT EXISTS parent_ticket_id UUID REFERENCES app.support_ticket(id) ON DELETE SET NULL
            """
        )
        await conn.execute(
            """
            CREATE INDEX IF NOT EXISTS idx_support_ticket_fingerprint
              ON app.support_ticket(fingerprint, created_at DESC)
              WHERE fingerprint <> ''
            """
        )
        await conn.execute(
            """
            CREATE INDEX IF NOT EXISTS idx_support_ticket_parent
              ON app.support_ticket(parent_ticket_id)
              WHERE parent_ticket_id IS NOT NULL
            """
        )
        await conn.execute(
            """
            CREATE TABLE IF NOT EXISTS app.learner_prepost_measure (
              id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
              learner_id UUID NOT NULL REFERENCES app.app_user(user_id) ON DELETE CASCADE,
              pilot_id TEXT NOT NULL DEFAULT 'phase3-pilot-draft',
              measure_name TEXT NOT NULL CHECK (
                measure_name IN ('self_efficacy','skill_proficiency','training_satisfaction')
              ),
              timepoint TEXT NOT NULL CHECK (timepoint IN ('pre','post')),
              raw_score NUMERIC(8,3) NOT NULL,
              min_score NUMERIC(8,3) NOT NULL DEFAULT 1.0,
              max_score NUMERIC(8,3) NOT NULL DEFAULT 5.0,
              instrument_version TEXT NOT NULL DEFAULT 'pilot-prepost-scaffold-2026-06-28',
              item_count INTEGER NOT NULL DEFAULT 1 CHECK (item_count >= 1 AND item_count <= 80),
              collected_at TIMESTAMPTZ NOT NULL DEFAULT now(),
              updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
              CHECK (max_score > min_score),
              CHECK (raw_score >= min_score AND raw_score <= max_score),
              UNIQUE (learner_id, pilot_id, measure_name, timepoint, instrument_version)
            )
            """
        )
        await conn.execute(
            """
            CREATE INDEX IF NOT EXISTS idx_learner_prepost_measure_pilot
              ON app.learner_prepost_measure(pilot_id, measure_name, timepoint)
            """
        )
        await conn.execute(
            """
            CREATE INDEX IF NOT EXISTS idx_learner_prepost_measure_learner
              ON app.learner_prepost_measure(learner_id, pilot_id)
            """
        )
        await conn.execute(
            """
            ALTER TABLE app.admin_health_event ENABLE ROW LEVEL SECURITY;
            DROP POLICY IF EXISTS p_admin_health_event_select ON app.admin_health_event;
            DROP POLICY IF EXISTS p_admin_health_event_insert ON app.admin_health_event;
            DROP POLICY IF EXISTS p_admin_health_event_delete ON app.admin_health_event;
            CREATE POLICY p_admin_health_event_select
              ON app.admin_health_event FOR SELECT
              USING (app.current_role_name() = 'admin');
            CREATE POLICY p_admin_health_event_insert
              ON app.admin_health_event FOR INSERT
              WITH CHECK (app.current_role_name() = 'admin');
            CREATE POLICY p_admin_health_event_delete
              ON app.admin_health_event FOR DELETE
              USING (app.current_role_name() = 'admin');

            ALTER TABLE app.admin_health_daily_rollup ENABLE ROW LEVEL SECURITY;
            DROP POLICY IF EXISTS p_admin_health_daily_rollup_select ON app.admin_health_daily_rollup;
            DROP POLICY IF EXISTS p_admin_health_daily_rollup_insert ON app.admin_health_daily_rollup;
            DROP POLICY IF EXISTS p_admin_health_daily_rollup_update ON app.admin_health_daily_rollup;
            DROP POLICY IF EXISTS p_admin_health_daily_rollup_delete ON app.admin_health_daily_rollup;
            ALTER TABLE app.admin_health_daily_rollup
              ADD COLUMN IF NOT EXISTS last_down_at TIMESTAMPTZ;
            CREATE POLICY p_admin_health_daily_rollup_select
              ON app.admin_health_daily_rollup FOR SELECT
              USING (app.current_role_name() = 'admin');
            CREATE POLICY p_admin_health_daily_rollup_insert
              ON app.admin_health_daily_rollup FOR INSERT
              WITH CHECK (app.current_role_name() = 'admin');
            CREATE POLICY p_admin_health_daily_rollup_update
              ON app.admin_health_daily_rollup FOR UPDATE
              USING (app.current_role_name() = 'admin')
              WITH CHECK (app.current_role_name() = 'admin');

            ALTER TABLE app.support_ticket ENABLE ROW LEVEL SECURITY;
            DROP POLICY IF EXISTS p_support_ticket_select ON app.support_ticket;
            DROP POLICY IF EXISTS p_support_ticket_insert ON app.support_ticket;
            DROP POLICY IF EXISTS p_support_ticket_update ON app.support_ticket;
            DROP POLICY IF EXISTS p_support_ticket_delete ON app.support_ticket;
            CREATE POLICY p_support_ticket_select
              ON app.support_ticket FOR SELECT
              USING (
                app.current_role_name() = 'admin'
                OR reporter_id = app.current_uid()
              );
            CREATE POLICY p_support_ticket_insert
              ON app.support_ticket FOR INSERT
              WITH CHECK (
                app.current_role_name() = 'admin'
                OR reporter_id = app.current_uid()
              );
            CREATE POLICY p_support_ticket_update
              ON app.support_ticket FOR UPDATE
              USING (app.current_role_name() = 'admin')
              WITH CHECK (app.current_role_name() = 'admin');
            CREATE POLICY p_support_ticket_delete
              ON app.support_ticket FOR DELETE
              USING (app.current_role_name() = 'admin');

            ALTER TABLE app.learner_prepost_measure ENABLE ROW LEVEL SECURITY;
            DROP POLICY IF EXISTS p_learner_prepost_measure_select ON app.learner_prepost_measure;
            DROP POLICY IF EXISTS p_learner_prepost_measure_insert ON app.learner_prepost_measure;
            DROP POLICY IF EXISTS p_learner_prepost_measure_update ON app.learner_prepost_measure;
            CREATE POLICY p_learner_prepost_measure_select
              ON app.learner_prepost_measure FOR SELECT
              USING (
                app.current_role_name() = 'admin'
                OR learner_id = app.current_uid()
                OR (
                  app.current_role_name() = 'instructor'
                  AND EXISTS (
                    SELECT 1 FROM app.app_user u
                    WHERE u.user_id = app.learner_prepost_measure.learner_id
                      AND u.cohort = current_setting('app.current_cohort', true)
                  )
                )
              );
            CREATE POLICY p_learner_prepost_measure_insert
              ON app.learner_prepost_measure FOR INSERT
              WITH CHECK (
                app.current_role_name() = 'admin'
                OR learner_id = app.current_uid()
              );
            CREATE POLICY p_learner_prepost_measure_update
              ON app.learner_prepost_measure FOR UPDATE
              USING (
                app.current_role_name() = 'admin'
                OR learner_id = app.current_uid()
              )
              WITH CHECK (
                app.current_role_name() = 'admin'
                OR learner_id = app.current_uid()
              );
            """
        )
        await conn.execute(
            """
            ALTER TABLE app.sessions
              ADD COLUMN IF NOT EXISTS runtime_case_id UUID,
              ADD COLUMN IF NOT EXISTS persona_code TEXT,
              ADD COLUMN IF NOT EXISTS persona_display_name TEXT,
              ADD COLUMN IF NOT EXISTS persona_difficulty TEXT,
              ADD COLUMN IF NOT EXISTS prev_rapport_credit REAL NOT NULL DEFAULT 0.0,
              ADD COLUMN IF NOT EXISTS session_goals JSONB NOT NULL DEFAULT '[]'::jsonb,
              ADD COLUMN IF NOT EXISTS learner_feedback_enabled BOOLEAN NOT NULL DEFAULT true
            """
        )
        await conn.execute(
            """
            ALTER TABLE app.session_state
              ADD COLUMN IF NOT EXISTS turns_in_stage INT NOT NULL DEFAULT 0
            """
        )
        await conn.execute(
            """
            ALTER TABLE app.turns
              ADD COLUMN IF NOT EXISTS audio_ref TEXT,
              ADD COLUMN IF NOT EXISTS silence_ms INT,
              ADD COLUMN IF NOT EXISTS speech_rate REAL,
              ADD COLUMN IF NOT EXISTS barge_in BOOLEAN,
              ADD COLUMN IF NOT EXISTS provider_events JSONB NOT NULL DEFAULT '[]'::jsonb
            """
        )
        await conn.execute(
            """
            INSERT INTO app.stage_def (stage_code, display_name, seq, base_openness)
            VALUES
              ('라포', '라포', 1, 0.15),
              ('탐색', '탐색', 2, 0.35),
              ('개입', '개입', 3, 0.55),
              ('정리', '정리', 4, 0.45)
            ON CONFLICT (stage_code) DO UPDATE SET
              display_name = EXCLUDED.display_name,
              seq = EXCLUDED.seq,
              base_openness = EXCLUDED.base_openness
            """
        )
        await conn.execute(
            """
            DROP POLICY IF EXISTS p_sessions_modify ON app.sessions;
            DROP POLICY IF EXISTS p_sessions_insert ON app.sessions;
            DROP POLICY IF EXISTS p_sessions_update ON app.sessions;
            DROP POLICY IF EXISTS p_sessions_delete ON app.sessions;

            CREATE POLICY p_sessions_insert ON app.sessions FOR INSERT WITH CHECK (
              app.is_ai_context() OR app.current_role_name() IN ('admin','instructor')
              OR learner_id = app.current_uid()
            );
            CREATE POLICY p_sessions_update ON app.sessions FOR UPDATE USING (
              app.is_ai_context() OR app.current_role_name() IN ('admin','instructor')
              OR learner_id = app.current_uid()
            ) WITH CHECK (
              app.is_ai_context() OR app.current_role_name() IN ('admin','instructor')
              OR learner_id = app.current_uid()
            );
            CREATE POLICY p_sessions_delete ON app.sessions FOR DELETE USING (
              app.is_ai_context() OR app.current_role_name() IN ('admin','instructor')
              OR learner_id = app.current_uid()
            )
            """
        )
        await conn.execute(
            """
            DROP POLICY IF EXISTS p_case_insert ON app.case_profile;
            DROP POLICY IF EXISTS p_case_update ON app.case_profile;

            CREATE POLICY p_case_insert ON app.case_profile FOR INSERT WITH CHECK (
              app.is_ai_context() OR app.current_role_name() IN ('admin','instructor')
              OR learner_id = app.current_uid()
            );
            CREATE POLICY p_case_update ON app.case_profile FOR UPDATE USING (
              app.is_ai_context() OR app.current_role_name() IN ('admin','instructor')
              OR learner_id = app.current_uid()
            ) WITH CHECK (
              app.is_ai_context() OR app.current_role_name() IN ('admin','instructor')
              OR learner_id = app.current_uid()
            )
            """
        )
        await conn.execute(
            """
            ALTER TABLE app.pinned_fact ENABLE ROW LEVEL SECURITY;
            DROP POLICY IF EXISTS p_pinned_insert ON app.pinned_fact;
            DROP POLICY IF EXISTS p_pinned_update ON app.pinned_fact;

            CREATE POLICY p_pinned_insert ON app.pinned_fact FOR INSERT WITH CHECK (
              app.current_role_name() IN ('admin','instructor')
              OR EXISTS (
                SELECT 1 FROM app.case_profile cp
                WHERE cp.case_id = app.pinned_fact.case_id
                  AND cp.learner_id = app.current_uid()
              )
            );
            CREATE POLICY p_pinned_update ON app.pinned_fact FOR UPDATE USING (
              app.current_role_name() IN ('admin','instructor')
              OR EXISTS (
                SELECT 1 FROM app.case_profile cp
                WHERE cp.case_id = app.pinned_fact.case_id
                  AND cp.learner_id = app.current_uid()
              )
            ) WITH CHECK (
              app.current_role_name() IN ('admin','instructor')
              OR EXISTS (
                SELECT 1 FROM app.case_profile cp
                WHERE cp.case_id = app.pinned_fact.case_id
                  AND cp.learner_id = app.current_uid()
              )
            )
            """
        )
        await conn.execute(
            """
            DROP POLICY IF EXISTS p_turns_modify ON app.turns;
            DROP POLICY IF EXISTS p_turns_insert ON app.turns;
            DROP POLICY IF EXISTS p_turns_update ON app.turns;
            DROP POLICY IF EXISTS p_turns_delete ON app.turns;

            CREATE POLICY p_turns_insert ON app.turns FOR INSERT WITH CHECK (
              app.is_ai_context()
              OR app.current_role_name() IN ('admin','instructor')
              OR EXISTS (
                SELECT 1 FROM app.sessions s
                WHERE s.id = app.turns.session_id
                  AND s.learner_id = app.current_uid()
              )
            );
            CREATE POLICY p_turns_update ON app.turns FOR UPDATE USING (
              app.is_ai_context()
              OR app.current_role_name() IN ('admin','instructor')
              OR EXISTS (
                SELECT 1 FROM app.sessions s
                WHERE s.id = app.turns.session_id
                  AND s.learner_id = app.current_uid()
              )
            ) WITH CHECK (
              app.is_ai_context()
              OR app.current_role_name() IN ('admin','instructor')
              OR EXISTS (
                SELECT 1 FROM app.sessions s
                WHERE s.id = app.turns.session_id
                  AND s.learner_id = app.current_uid()
              )
            );
            CREATE POLICY p_turns_delete ON app.turns FOR DELETE USING (
              app.is_ai_context()
              OR app.current_role_name() IN ('admin','instructor')
              OR EXISTS (
                SELECT 1 FROM app.sessions s
                WHERE s.id = app.turns.session_id
                  AND s.learner_id = app.current_uid()
              )
            )
            """
        )
        if not await _runtime_tables_ready(conn):
            raise RuntimeError(
                "auth/user development schema bootstrap did not satisfy readiness"
            )


def _managed_user_from_row(row) -> ManagedUser:
    return ManagedUser(
        user_id=str(row["user_id"]),
        email=row["email"],
        display_name=row["display_name"] or row["email"],
        role=_app_role(row["role"]),
        admin_access=bool(_row_value(row, "admin_access", False)),
        account_status=_account_status(_row_value(row, "account_status", "approved")),
        cohort_ids=_cohort_ids(_row_value(row, "cohort")),
        affiliation=row["affiliation"] or DEFAULT_AFFILIATION,
        legal_name=_row_value(row, "legal_name", "") or "",
        department=_row_value(row, "department", "") or "",
        grade_level=_row_value(row, "grade_level", "") or "",
        phone=_row_value(row, "phone", "") or "",
        contact_address=_row_value(row, "contact_address", "") or "",
        nickname=_row_value(row, "nickname", "") or "",
        self_introduction=_row_value(row, "self_introduction", "") or "",
        avatar_url=_row_value(row, "avatar_url", "") or "",
        consent_at=_optional_ts(_row_value(row, "consent_at")),
        profile_completed_at=_optional_ts(_row_value(row, "profile_completed_at")),
        terms_agreed_at=_optional_ts(_row_value(row, "terms_agreed_at")),
        privacy_agreed_at=_optional_ts(_row_value(row, "privacy_agreed_at")),
        terms_version=_row_value(row, "terms_version", "") or "",
        privacy_version=_row_value(row, "privacy_version", "") or "",
        created_at=_ts(row["created_at"]),
        last_seen_at=_ts(row["last_seen_at"]),
        learner_feedback_enabled=bool(
            _row_value(row, "learner_feedback_enabled", True)
        ),
    )


def _memory_upsert_managed_user(data: ManagedUserMemoryInput) -> ManagedUser:
    now = time.time()
    normalized_email = _normalize_email(data.email)
    if normalized_email in _inactive_emails and not data.reactivate:
        raise InactiveUserError("user is inactive")
    if data.reactivate:
        _inactive_emails.discard(normalized_email)
    uid = (
        data.user_id
        or _email_index.get(normalized_email)
        or user_id_from_email(normalized_email)
    )
    current = _users.get(uid)
    user = ManagedUser(
        user_id=uid,
        email=normalized_email,
        display_name=(data.display_name.strip() if data.display_name else "")
        or normalized_email,
        role=data.role,
        admin_access=data.admin_access
        if data.admin_access is not None
        else (current.admin_access if current else False),
        account_status=(
            "suspended"
            if current is not None
            and current.account_status == "suspended"
            and not data.reactivate
            else data.account_status
            or (current.account_status if current else "approved")
        ),
        cohort_ids=(
            list(data.cohort_ids)
            if data.cohort_ids is not None
            else (list(current.cohort_ids) if current else [])
        ),
        affiliation=(
            data.affiliation.strip()
            if data.affiliation
            else (current.affiliation if current else DEFAULT_AFFILIATION)
        ),
        legal_name=(
            data.legal_name.strip()
            if data.legal_name is not None
            else (current.legal_name if current else "")
        ),
        department=(
            data.department.strip()
            if data.department is not None
            else (current.department if current else "")
        ),
        grade_level=(
            data.grade_level.strip()
            if data.grade_level is not None
            else (current.grade_level if current else "")
        ),
        phone=data.phone.strip()
        if data.phone is not None
        else (current.phone if current else ""),
        contact_address=(
            data.contact_address.strip()
            if data.contact_address is not None
            else (current.contact_address if current else "")
        ),
        nickname=data.nickname.strip()
        if data.nickname is not None
        else (current.nickname if current else ""),
        self_introduction=(
            data.self_introduction.strip()
            if data.self_introduction is not None
            else (current.self_introduction if current else "")
        ),
        avatar_url=data.avatar_url.strip()
        if data.avatar_url is not None
        else (current.avatar_url if current else ""),
        consent_at=data.consent_at
        if data.consent_at is not None
        else (current.consent_at if current else None),
        profile_completed_at=(
            data.profile_completed_at
            if data.profile_completed_at is not None
            else (current.profile_completed_at if current else None)
        ),
        terms_agreed_at=(
            data.terms_agreed_at
            if data.terms_agreed_at is not None
            else (current.terms_agreed_at if current else None)
        ),
        privacy_agreed_at=(
            data.privacy_agreed_at
            if data.privacy_agreed_at is not None
            else (current.privacy_agreed_at if current else None)
        ),
        terms_version=(
            data.terms_version.strip()
            if data.terms_version is not None
            else (current.terms_version if current else "")
        ),
        privacy_version=(
            data.privacy_version.strip()
            if data.privacy_version is not None
            else (current.privacy_version if current else "")
        ),
        created_at=current.created_at if current else now,
        last_seen_at=now,
        learner_feedback_enabled=(
            data.learner_feedback_enabled
            if data.learner_feedback_enabled is not None
            else (current.learner_feedback_enabled if current else True)
        ),
    )
    _users[uid] = user
    _email_index[normalized_email] = uid
    return user


def _sync_managed_user_sessions(user: ManagedUser) -> None:
    """관리 사용자 변경을 현재 로그인 세션의 권한 스냅샷에 반영한다."""
    for session in _sessions.values():
        if session.user_id != user.user_id:
            continue
        session.display_name = user.display_name
        session.role = user.role
        session.admin_access = has_admin_access(
            user.email,
            user.role,
            user.admin_access,
        )
        session.super_admin = is_super_admin_email(user.email)
        session.account_status = user.account_status
        session.cohort_ids = list(user.cohort_ids)
        session.consent_at = user.consent_at
        session.profile_completed_at = user.profile_completed_at
        session.learner_feedback_enabled = user.learner_feedback_enabled


async def upsert_managed_user(data: ManagedUserUpsertInput) -> ManagedUser:
    normalized_email = _normalize_email(data.email)
    normalized_external_id = _normalize_external_id(data.external_id, normalized_email)
    manual_external_id = f"email:{normalized_email}"
    desired_admin_access = (
        data.admin_access
        if data.admin_access is not None
        else (
            True
            if has_admin_access(normalized_email, data.role, False)
            else None
        )
    )
    desired_account_status = data.account_status or _initial_account_status(
        email=normalized_email,
        external_id=normalized_external_id,
        reactivate=data.reactivate,
    )
    try:
        pool = get_pool()
        async with pool.acquire() as conn:
            if data.user_id is None and normalized_external_id != manual_external_id:
                row = await conn.fetchrow(
                    """
                UPDATE app.app_user SET
                  external_id = $1,
                  display_name = COALESCE(NULLIF(display_name, ''), $3),
                  cohort = COALESCE(cohort, $4),
                  affiliation = COALESCE(NULLIF(affiliation, ''), $5),
                  admin_access = CASE
                    WHEN $7::boolean IS NULL THEN admin_access
                    ELSE $7::boolean
                  END,
                  account_status = CASE
                    WHEN account_status = 'suspended' THEN 'suspended'
                    WHEN $8::text = 'approved' THEN 'approved'
                    ELSE account_status
                  END,
                  last_seen_at = now(),
                  updated_at = now()
                WHERE user_id = (
                  SELECT user_id
                  FROM app.app_user
                  WHERE lower(email) = $2
                    AND is_active
                    AND (external_id = $6 OR external_id IS NULL)
                    AND NOT EXISTS (
                      SELECT 1 FROM app.app_user existing
                      WHERE existing.external_id = $1
                    )
                  ORDER BY last_seen_at DESC
                  LIMIT 1
                )
                RETURNING
                  user_id,
                  email,
                  display_name,
                  role,
                  admin_access,
                  learner_feedback_enabled,
                  account_status,
                  cohort,
                  affiliation,
                  legal_name,
                  department,
                  grade_level,
                  phone,
                  contact_address,
                  nickname,
                  self_introduction,
                  avatar_url,
                  consent_at,
                  profile_completed_at,
                  terms_agreed_at,
                  privacy_agreed_at,
                  terms_version,
                  privacy_version,
                  created_at,
                  last_seen_at
                """,
                    normalized_external_id,
                    normalized_email,
                    (
                        data.display_name.strip()
                        if data.display_name
                        else normalized_email
                    ),
                    _cohort_value(data.cohort_ids),
                    data.affiliation or DEFAULT_AFFILIATION,
                    manual_external_id,
                    desired_admin_access,
                    desired_account_status,
                )
                if row is not None:
                    user = _managed_user_from_row(row)
                    _memory_upsert_managed_user(
                        ManagedUserMemoryInput.from_user(user, reactivate=True)
                    )
                    return user
            row = await conn.fetchrow(
                """
            INSERT INTO app.app_user (
              external_id,
              email,
              display_name,
              role,
              admin_access,
              learner_feedback_enabled,
              cohort,
              affiliation,
              account_status,
              last_seen_at,
              updated_at
            )
            VALUES (
              $1, $2, $3, $4, COALESCE($9::boolean, false),
              COALESCE($10::boolean, true), $5, $6, $8, now(), now()
            )
            ON CONFLICT (external_id) DO UPDATE SET
              email = EXCLUDED.email,
              display_name = COALESCE(NULLIF(EXCLUDED.display_name, ''), app.app_user.display_name),
              role = EXCLUDED.role,
              admin_access = CASE
                WHEN $9::boolean IS NULL THEN app.app_user.admin_access
                ELSE EXCLUDED.admin_access
              END,
              learner_feedback_enabled = CASE
                WHEN $10::boolean IS NULL THEN app.app_user.learner_feedback_enabled
                ELSE EXCLUDED.learner_feedback_enabled
              END,
              cohort = COALESCE(EXCLUDED.cohort, app.app_user.cohort),
              affiliation = COALESCE(NULLIF(EXCLUDED.affiliation, ''), app.app_user.affiliation),
              is_active = CASE WHEN $7 THEN TRUE ELSE app.app_user.is_active END,
              account_status = CASE
                WHEN app.app_user.account_status = 'suspended' THEN 'suspended'
                WHEN EXCLUDED.account_status = 'approved' THEN 'approved'
                WHEN app.app_user.account_status = 'approved' THEN 'approved'
                ELSE EXCLUDED.account_status
              END,
              last_seen_at = now(),
              updated_at = now()
            WHERE app.app_user.is_active OR $7
            RETURNING
              user_id,
              email,
              display_name,
              role,
              admin_access,
              learner_feedback_enabled,
              account_status,
              cohort,
              affiliation,
              legal_name,
              department,
              grade_level,
              phone,
              contact_address,
              nickname,
              self_introduction,
              avatar_url,
              consent_at,
              profile_completed_at,
              terms_agreed_at,
              privacy_agreed_at,
              terms_version,
              privacy_version,
              created_at,
              last_seen_at
            """,
                normalized_external_id,
                normalized_email,
                (data.display_name.strip() if data.display_name else normalized_email),
                _db_role(data.role),
                _cohort_value(data.cohort_ids),
                data.affiliation or DEFAULT_AFFILIATION,
                data.reactivate,
                desired_account_status,
                desired_admin_access,
                data.learner_feedback_enabled,
            )
            if row is None:
                _inactive_emails.add(normalized_email)
                raise InactiveUserError("user is inactive")
            user = _managed_user_from_row(row)
            _memory_upsert_managed_user(
                ManagedUserMemoryInput.from_user(user, reactivate=True)
            )
            return user
    except InactiveUserError:
        raise
    except Exception:
        if not runtime_fallback_allowed():
            logger.exception("managed user persistence failed")
        require_runtime_fallback_allowed("managed user")
        current = _users.get(data.user_id or "") or _users.get(
            _email_index.get(normalized_email, "")
        )
        fallback_uid = data.user_id or (
            current.user_id
            if current is not None
            else user_id_from_external_id(normalized_external_id)
        )
        fallback_account_status = desired_account_status
        if current is not None:
            if current.account_status == "suspended":
                fallback_account_status = "suspended"
            elif (
                current.account_status == "approved"
                and desired_account_status == "pending"
            ):
                fallback_account_status = "approved"
        return _memory_upsert_managed_user(
            ManagedUserMemoryInput(
                email=normalized_email,
                display_name=data.display_name,
                role=data.role,
                admin_access=desired_admin_access,
                account_status=fallback_account_status,
                cohort_ids=data.cohort_ids,
                user_id=fallback_uid,
                affiliation=data.affiliation,
                learner_feedback_enabled=data.learner_feedback_enabled,
                reactivate=data.reactivate,
            )
        )


_MANAGED_USER_SELECT_SQL = """
SELECT
  user_id,
  email,
  display_name,
  role,
  admin_access,
  learner_feedback_enabled,
  account_status,
  cohort,
  affiliation,
  legal_name,
  department,
  grade_level,
  phone,
  contact_address,
  nickname,
  self_introduction,
  avatar_url,
  consent_at,
  profile_completed_at,
  terms_agreed_at,
  privacy_agreed_at,
  terms_version,
  privacy_version,
  created_at,
  last_seen_at
FROM app.app_user
"""


async def get_managed_user(user_id: str) -> ManagedUser | None:
    try:
        pool = get_pool()
        async with pool.acquire() as conn:
            row = await conn.fetchrow(
                f"""
                {_MANAGED_USER_SELECT_SQL}
                WHERE user_id = $1::uuid AND is_active
                """,
                user_id,
            )
            if row is not None:
                return _managed_user_from_row(row)
    except Exception:
        require_runtime_fallback_allowed("managed user")
    if not runtime_fallback_allowed():
        return None
    return _users.get(user_id)


async def get_managed_user_by_email(email: str) -> ManagedUser | None:
    normalized_email = _normalize_email(email)
    if not normalized_email:
        return None
    manual_external_id = f"email:{normalized_email}"
    try:
        pool = get_pool()
        async with pool.acquire() as conn:
            row = await conn.fetchrow(
                f"""
                {_MANAGED_USER_SELECT_SQL}
                WHERE lower(email) = $1 AND is_active
                ORDER BY
                  CASE WHEN external_id = $2 THEN 0 ELSE 1 END,
                  last_seen_at DESC
                LIMIT 1
                """,
                normalized_email,
                manual_external_id,
            )
            if row is not None:
                return _managed_user_from_row(row)
    except Exception:
        require_runtime_fallback_allowed("managed user lookup")
    if not runtime_fallback_allowed():
        return None
    uid = _email_index.get(normalized_email)
    return _users.get(uid or "")


async def get_managed_user_by_auth_alias(external_id: str) -> ManagedUser | None:
    """Resolve an explicitly approved provider identity to its canonical user.

    Email equality never creates an alias. A configured alias whose canonical
    target is inactive or suspended fails closed instead of creating a new user.
    """

    normalized_external_id = (external_id or "").strip().lower()
    if not normalized_external_id:
        return None
    try:
        pool = get_pool()
        async with pool.acquire() as conn:
            row = await conn.fetchrow(
                """
                SELECT
                  u.user_id,
                  u.email,
                  u.display_name,
                  u.role,
                  u.admin_access,
                  u.learner_feedback_enabled,
                  u.account_status,
                  u.cohort,
                  u.affiliation,
                  u.legal_name,
                  u.department,
                  u.grade_level,
                  u.phone,
                  u.contact_address,
                  u.nickname,
                  u.self_introduction,
                  u.avatar_url,
                  u.consent_at,
                  u.profile_completed_at,
                  u.terms_agreed_at,
                  u.privacy_agreed_at,
                  u.terms_version,
                  u.privacy_version,
                  u.created_at,
                  u.last_seen_at,
                  u.is_active
                FROM app.auth_identity_alias AS alias
                JOIN app.app_user AS u ON u.user_id = alias.user_id
                WHERE alias.external_id = $1
                """,
                normalized_external_id,
            )
            if row is None:
                return None
            if not bool(row["is_active"]) or _account_status(row["account_status"]) == "suspended":
                raise InactiveUserError("aliased user is inactive")
            user = _managed_user_from_row(row)
            _memory_upsert_managed_user(
                ManagedUserMemoryInput.from_user(user, reactivate=True)
            )
            _auth_identity_alias_index[normalized_external_id] = user.user_id
            return user
    except InactiveUserError:
        raise
    except Exception:
        require_runtime_fallback_allowed("auth identity alias lookup")

    uid = _auth_identity_alias_index.get(normalized_external_id)
    user = _users.get(uid or "")
    if user is None:
        return None
    if user.account_status == "suspended" or user.email in _inactive_emails:
        raise InactiveUserError("aliased user is inactive")
    return user


async def list_managed_users() -> tuple[list[ManagedUser], bool]:
    try:
        pool = get_pool()
        async with pool.acquire() as conn:
            rows = await conn.fetch(
                """
                SELECT
                  user_id,
                  email,
                  display_name,
                  role,
                  admin_access,
                  learner_feedback_enabled,
                  account_status,
                  cohort,
                  affiliation,
                  legal_name,
                  department,
                  grade_level,
                  phone,
                  contact_address,
                  nickname,
                  self_introduction,
                  avatar_url,
                  consent_at,
                  profile_completed_at,
                  terms_agreed_at,
                  privacy_agreed_at,
                  terms_version,
                  privacy_version,
                  created_at,
                  last_seen_at
                FROM app.app_user
                WHERE is_active
                ORDER BY last_seen_at DESC
                """
            )
            return [_managed_user_from_row(row) for row in rows], True
    except Exception:
        require_runtime_fallback_allowed("managed user list")
    return sorted(_users.values(), key=lambda u: u.last_seen_at, reverse=True), False


async def update_managed_user(
    user_id: str,
    patch: ManagedUserPatch,
) -> ManagedUser | None:
    try:
        pool = get_pool()
        async with pool.acquire() as conn:
            row = await conn.fetchrow(
                """
                UPDATE app.app_user SET
                  display_name = COALESCE($2, display_name),
                  role = COALESCE($3, role),
                  account_status = COALESCE($18, account_status),
                  admin_access = COALESCE($19, admin_access),
                  learner_feedback_enabled = COALESCE($20, learner_feedback_enabled),
                  cohort = CASE WHEN $4 THEN $5 ELSE cohort END,
                  affiliation = COALESCE($6, affiliation),
                  legal_name = COALESCE($7, legal_name),
                  department = COALESCE($8, department),
                  grade_level = COALESCE($9, grade_level),
                  phone = COALESCE($10, phone),
                  contact_address = COALESCE($11, contact_address),
                  nickname = COALESCE($12, nickname),
                  self_introduction = COALESCE($13, self_introduction),
                  avatar_url = COALESCE($14, avatar_url),
                  profile_completed_at = CASE WHEN $15 THEN now() ELSE profile_completed_at END,
                  terms_agreed_at = CASE WHEN $15 THEN now() ELSE terms_agreed_at END,
                  privacy_agreed_at = CASE WHEN $15 THEN now() ELSE privacy_agreed_at END,
                  terms_version = CASE WHEN $15 THEN COALESCE($16, terms_version) ELSE terms_version END,
                  privacy_version = CASE WHEN $15 THEN COALESCE($17, privacy_version) ELSE privacy_version END,
                  updated_at = now(),
                  last_seen_at = now()
                WHERE user_id = $1::uuid AND is_active
                RETURNING
                  user_id,
                  email,
                  display_name,
                  role,
                  admin_access,
                  learner_feedback_enabled,
                  account_status,
                  cohort,
                  affiliation,
                  legal_name,
                  department,
                  grade_level,
                  phone,
                  contact_address,
                  nickname,
                  self_introduction,
                  avatar_url,
                  consent_at,
                  profile_completed_at,
                  terms_agreed_at,
                  privacy_agreed_at,
                  terms_version,
                  privacy_version,
                  created_at,
                  last_seen_at
                """,
                user_id,
                patch.display_name.strip() if patch.display_name is not None else None,
                _db_role(patch.role) if patch.role is not None else None,
                patch.cohort_ids is not None,
                _cohort_value(patch.cohort_ids),
                patch.affiliation.strip() if patch.affiliation is not None else None,
                patch.legal_name.strip() if patch.legal_name is not None else None,
                patch.department.strip() if patch.department is not None else None,
                patch.grade_level.strip() if patch.grade_level is not None else None,
                patch.phone.strip() if patch.phone is not None else None,
                patch.contact_address.strip()
                if patch.contact_address is not None
                else None,
                patch.nickname.strip() if patch.nickname is not None else None,
                patch.self_introduction.strip()
                if patch.self_introduction is not None
                else None,
                patch.avatar_url.strip() if patch.avatar_url is not None else None,
                patch.complete_onboarding,
                patch.terms_version.strip()
                if patch.terms_version is not None
                else None,
                patch.privacy_version.strip()
                if patch.privacy_version is not None
                else None,
                patch.account_status,
                patch.admin_access,
                patch.learner_feedback_enabled,
            )
            if row is not None:
                next_user = _managed_user_from_row(row)
                _memory_upsert_managed_user(ManagedUserMemoryInput.from_user(next_user))
                _sync_managed_user_sessions(next_user)
                return next_user
    except Exception:
        require_runtime_fallback_allowed("managed user update")

    if not runtime_fallback_allowed():
        return None
    current = _users.get(user_id)
    if current is None:
        return None

    next_user = ManagedUser(
        user_id=current.user_id,
        email=current.email,
        display_name=patch.display_name.strip()
        if patch.display_name is not None
        else current.display_name,
        role=patch.role if patch.role is not None else current.role,
        admin_access=patch.admin_access
        if patch.admin_access is not None
        else current.admin_access,
        account_status=patch.account_status
        if patch.account_status is not None
        else current.account_status,
        learner_feedback_enabled=patch.learner_feedback_enabled
        if patch.learner_feedback_enabled is not None
        else current.learner_feedback_enabled,
        cohort_ids=list(patch.cohort_ids)
        if patch.cohort_ids is not None
        else current.cohort_ids,
        affiliation=patch.affiliation.strip()
        if patch.affiliation is not None
        else current.affiliation,
        legal_name=patch.legal_name.strip()
        if patch.legal_name is not None
        else current.legal_name,
        department=patch.department.strip()
        if patch.department is not None
        else current.department,
        grade_level=patch.grade_level.strip()
        if patch.grade_level is not None
        else current.grade_level,
        phone=patch.phone.strip() if patch.phone is not None else current.phone,
        contact_address=(
            patch.contact_address.strip()
            if patch.contact_address is not None
            else current.contact_address
        ),
        nickname=patch.nickname.strip()
        if patch.nickname is not None
        else current.nickname,
        self_introduction=(
            patch.self_introduction.strip()
            if patch.self_introduction is not None
            else current.self_introduction
        ),
        avatar_url=patch.avatar_url.strip()
        if patch.avatar_url is not None
        else current.avatar_url,
        consent_at=current.consent_at,
        profile_completed_at=time.time()
        if patch.complete_onboarding
        else current.profile_completed_at,
        terms_agreed_at=time.time()
        if patch.complete_onboarding
        else current.terms_agreed_at,
        privacy_agreed_at=time.time()
        if patch.complete_onboarding
        else current.privacy_agreed_at,
        terms_version=(
            patch.terms_version.strip()
            if patch.complete_onboarding and patch.terms_version is not None
            else current.terms_version
        ),
        privacy_version=(
            patch.privacy_version.strip()
            if patch.complete_onboarding and patch.privacy_version is not None
            else current.privacy_version
        ),
        created_at=current.created_at,
        last_seen_at=time.time(),
    )
    _users[user_id] = next_user
    _sync_managed_user_sessions(next_user)
    return next_user


async def record_user_consent(user_id: str) -> float | None:
    """Mark a learner consent receipt timestamp for the current user."""
    try:
        pool = get_pool()
        async with pool.acquire() as conn:
            row = await conn.fetchrow(
                """
                UPDATE app.app_user SET
                  consent_at = now(),
                  updated_at = now(),
                  last_seen_at = now()
                WHERE user_id = $1::uuid AND is_active
                RETURNING consent_at
                """,
                user_id,
            )
            if row is not None:
                consent_at = _optional_ts(row["consent_at"])
                memory_user = _users.get(user_id)
                if memory_user is not None:
                    _users[user_id] = replace(
                        memory_user,
                        consent_at=consent_at,
                        last_seen_at=time.time(),
                    )
                for session in _sessions.values():
                    if session.user_id == user_id:
                        session.consent_at = consent_at
                return consent_at
    except Exception:
        require_runtime_fallback_allowed("consent update")

    if not runtime_fallback_allowed():
        return None
    current = _users.get(user_id)
    if current is None:
        return None
    consent_at = time.time()
    _users[user_id] = replace(current, consent_at=consent_at, last_seen_at=consent_at)
    for session in _sessions.values():
        if session.user_id == user_id:
            session.consent_at = consent_at
    return consent_at


async def withdraw_user_consent(user_id: str) -> bool:
    """Clear consent and revoke practice eligibility until the user consents again."""
    changed = False
    try:
        pool = get_pool()
        async with pool.acquire() as conn:
            result = await conn.execute(
                """
                UPDATE app.app_user SET
                  consent_at = NULL,
                  updated_at = now(),
                  last_seen_at = now()
                WHERE user_id = $1::uuid AND is_active
                """,
                user_id,
            )
            changed = result.endswith(" 1")
    except Exception:
        require_runtime_fallback_allowed("consent withdrawal")

    if not runtime_fallback_allowed():
        return changed
    current = _users.get(user_id)
    if current is not None:
        _users[user_id] = replace(current, consent_at=None, last_seen_at=time.time())
        changed = True
    for session in _sessions.values():
        if session.user_id == user_id:
            session.consent_at = None
    return changed


async def user_has_consent(user_id: str) -> bool:
    managed = await get_managed_user(user_id)
    return bool(managed and managed.consent_at is not None)


async def user_onboarding_complete(user_id: str) -> bool:
    managed = await get_managed_user(user_id)
    return bool(
        managed
        and managed.profile_completed_at is not None
        and managed.terms_agreed_at is not None
        and managed.privacy_agreed_at is not None
        and bool(managed.nickname.strip())
        and bool(managed.self_introduction.strip())
    )


async def deactivate_managed_user(user_id: str) -> bool:
    """Deactivate a managed user and revoke their active browser sessions."""
    changed = False
    inactive_email: str | None = None
    try:
        pool = get_pool()
        async with pool.acquire() as conn:
            row = await conn.fetchrow(
                "SELECT email FROM app.app_user WHERE user_id = $1::uuid",
                user_id,
            )
            if row is not None and row["email"]:
                inactive_email = _normalize_email(row["email"])
            result = await conn.execute(
                """
                UPDATE app.app_user SET
                  is_active = FALSE,
                  updated_at = now(),
                  last_seen_at = now()
                WHERE user_id = $1::uuid AND is_active
                """,
                user_id,
            )
            changed = result.endswith(" 1")
            await conn.execute(
                "UPDATE app.auth_session SET revoked_at = now() WHERE user_id = $1::uuid",
                user_id,
            )
    except Exception:
        require_runtime_fallback_allowed("managed user deactivation")

    if not runtime_fallback_allowed():
        return changed
    memory_user = _users.pop(user_id, None)
    if memory_user is not None:
        inactive_email = _normalize_email(memory_user.email)
    if inactive_email:
        _inactive_emails.add(inactive_email)
    for email, uid in list(_email_index.items()):
        if uid == user_id:
            _email_index.pop(email, None)
    for sid_hash, session in list(_sessions.items()):
        if session.user_id == user_id:
            _sessions.pop(sid_hash, None)
    return changed or memory_user is not None


async def active_session_count(user_id: str) -> int:
    try:
        pool = get_pool()
        async with pool.acquire() as conn:
            return int(
                await conn.fetchval(
                    """
                    SELECT count(*)
                    FROM app.auth_session
                    WHERE user_id = $1::uuid
                      AND revoked_at IS NULL
                      AND expires_at > now()
                    """,
                    user_id,
                )
                or 0
            )
    except Exception:
        require_runtime_fallback_allowed("auth session count")
    now = time.time()
    count = 0
    for session in _sessions.values():
        if session.user_id == user_id and session.expires_at > now:
            count += 1
    return count


async def create_session(
    *,
    email: str,
    display_name: str,
    role: RoleName,
    cohort_ids: list[str] | None = None,
    user_id: str | None = None,
    external_id: str | None = None,
    account_status: AccountStatus | None = None,
    managed_user: ManagedUser | None = None,
) -> tuple[str, SessionUser]:
    raw_sid = secrets.token_urlsafe(32)
    normalized_email = _normalize_email(email)
    managed = managed_user
    if managed is None:
        managed = await upsert_managed_user(
            ManagedUserUpsertInput(
                email=normalized_email,
                display_name=display_name,
                role=role,
                cohort_ids=cohort_ids,
                user_id=user_id,
                external_id=external_id,
                account_status=account_status,
                reactivate=False,
            )
        )
    if managed.account_status == "pending":
        try:
            await notifications.enqueue_account_pending_approval(
                user_id=managed.user_id,
                email=managed.email,
                display_name=managed.display_name,
                role=managed.role,
            )
        except Exception as exc:
            logger.warning("account approval notification enqueue failed: %s", exc)
    expires_at = time.time() + settings.session_ttl_seconds
    user = SessionUser(
        user_id=managed.user_id,
        email=normalized_email,
        display_name=managed.display_name,
        role=managed.role,
        admin_access=has_admin_access(
            managed.email, managed.role, managed.admin_access
        ),
        super_admin=is_super_admin_email(managed.email),
        account_status=managed.account_status,
        cohort_ids=list(managed.cohort_ids),
        consent_at=managed.consent_at,
        profile_completed_at=managed.profile_completed_at,
        expires_at=expires_at,
        learner_feedback_enabled=managed.learner_feedback_enabled,
    )
    sid_hash = _sid_hash(raw_sid)
    try:
        pool = get_pool()
        async with pool.acquire() as conn:
            await conn.execute(
                """
                INSERT INTO app.auth_session (
                  sid_hash, user_id, role, display_name, cohort_ids,
                  login_email, expires_at, last_seen_at
                )
                VALUES ($1, $2::uuid, $3, $4, $5::jsonb, $6, $7, now())
                ON CONFLICT (sid_hash) DO UPDATE SET
                  user_id = EXCLUDED.user_id,
                  role = EXCLUDED.role,
                  display_name = EXCLUDED.display_name,
                  cohort_ids = EXCLUDED.cohort_ids,
                  login_email = EXCLUDED.login_email,
                  expires_at = EXCLUDED.expires_at,
                  revoked_at = NULL,
                  last_seen_at = now()
                """,
                sid_hash,
                managed.user_id,
                managed.role,
                managed.display_name,
                list(managed.cohort_ids),
                normalized_email,
                datetime.fromtimestamp(expires_at, tz=timezone.utc),
            )
    except Exception:
        require_runtime_fallback_allowed("browser session")
    _sessions[sid_hash] = user
    return raw_sid, user


async def get_session(raw_sid: str | None) -> SessionUser | None:
    if not raw_sid:
        return None
    key = _sid_hash(raw_sid)
    try:
        pool = get_pool()
        async with pool.acquire() as conn:
            row = await conn.fetchrow(
                """
                SELECT
                  s.expires_at,
                  u.user_id,
                  COALESCE(NULLIF(s.login_email, ''), u.email) AS login_email,
                  u.email AS canonical_email,
                  COALESCE(u.display_name, s.display_name, u.email) AS display_name,
                  u.role,
                  u.admin_access,
                  u.learner_feedback_enabled,
                  u.account_status,
                  u.cohort,
                  u.consent_at,
                  u.profile_completed_at
                FROM app.auth_session s
                JOIN app.app_user u ON u.user_id = s.user_id
                WHERE s.sid_hash = $1
                  AND s.revoked_at IS NULL
                  AND s.expires_at > now()
                  AND u.is_active
                """,
                key,
            )
            if row is not None:
                app_role = _app_role(row["role"])
                stored_admin_access = bool(
                    _row_value(row, "admin_access", False)
                )
                effective_admin_access = has_admin_access(
                    row["canonical_email"],
                    app_role,
                    stored_admin_access,
                )
                await conn.execute(
                    "UPDATE app.auth_session SET last_seen_at = now() WHERE sid_hash = $1",
                    key,
                )
                await conn.execute(
                    """
                    UPDATE app.app_user SET
                      last_seen_at = now(),
                      admin_access = CASE
                        WHEN $2 THEN TRUE
                        ELSE admin_access
                      END,
                      updated_at = CASE
                        WHEN $2 AND NOT admin_access THEN now()
                        ELSE updated_at
                      END
                    WHERE user_id = $1
                    """,
                    row["user_id"],
                    effective_admin_access,
                )
                return SessionUser(
                    user_id=str(row["user_id"]),
                    email=row["login_email"],
                    display_name=row["display_name"],
                    role=app_role,
                    admin_access=effective_admin_access,
                    super_admin=is_super_admin_email(row["canonical_email"]),
                    account_status=_account_status(row["account_status"]),
                    cohort_ids=_cohort_ids(row["cohort"]),
                    consent_at=_optional_ts(row["consent_at"]),
                    profile_completed_at=_optional_ts(row["profile_completed_at"]),
                    expires_at=_ts(row["expires_at"]),
                    learner_feedback_enabled=bool(
                        _row_value(row, "learner_feedback_enabled", True)
                    ),
                )
    except Exception:
        require_runtime_fallback_allowed("browser session")

    if not runtime_fallback_allowed():
        return None
    user = _sessions.get(key)
    if user is None:
        return None
    if user.expires_at <= time.time():
        _sessions.pop(key, None)
        return None
    managed = _users.get(user.user_id)
    if managed is not None:
        managed.last_seen_at = time.time()
        user.display_name = managed.display_name
        user.role = managed.role
        user.admin_access = has_admin_access(
            managed.email, managed.role, managed.admin_access
        )
        user.super_admin = is_super_admin_email(managed.email)
        user.account_status = managed.account_status
        user.cohort_ids = list(managed.cohort_ids)
        user.consent_at = managed.consent_at
        user.profile_completed_at = managed.profile_completed_at
        user.learner_feedback_enabled = managed.learner_feedback_enabled
    return user


async def revoke_session(raw_sid: str | None) -> None:
    if not raw_sid:
        return
    key = _sid_hash(raw_sid)
    _sessions.pop(key, None)
    try:
        pool = get_pool()
        async with pool.acquire() as conn:
            await conn.execute(
                "UPDATE app.auth_session SET revoked_at = now() WHERE sid_hash = $1",
                key,
            )
    except Exception:
        require_runtime_fallback_allowed("browser session revoke")