From 9c9181bc505198db3d9a861db02c268ecfa4a6aa Mon Sep 17 00:00:00 2001
From: febux <deadcove@mail.ru>
Date: Thu, 13 Aug 2026 00:08:51 +0300
Subject: [PATCH] New notification service was added. Auth routes were
 extended.

---
 .env.example                                  |  16 +
 Makefile                                      |  11 +-
 docker-compose.yml                            |  42 ++
 docs/ARCHITECTURE.md                          |  14 +-
 migrations/versions/0003_web_auth.py          |  56 +++
 pyproject.toml                                |   9 +
 src/contract_check/api/app.py                 |  19 +-
 src/contract_check/api/deps.py                | 109 ++++-
 src/contract_check/api/routes/README.md       |  97 +++++
 src/contract_check/api/routes/auth.py         | 399 +++++++++++++++++-
 src/contract_check/core/auth.py               |  84 ++++
 src/contract_check/core/auth_refresh.py       |  58 +++
 src/contract_check/core/auth_refresh_key.py   |  15 +
 src/contract_check/core/config.py             |  26 ++
 src/contract_check/core/db/models.py          |  16 +-
 src/contract_check/core/metrics.py            |   6 +-
 src/contract_check/core/mq/consumer.py        |  19 +-
 src/contract_check/core/mq/messages.py        |  31 +-
 src/contract_check/core/mq/topology.py        |  35 +-
 .../core/notifications/__init__.py            |   0
 .../core/notifications/publisher.py           |  86 ++++
 .../core/notifications/transport.py           |  88 ++++
 src/contract_check/core/security/__init__.py  |   0
 src/contract_check/core/security/passwords.py |  55 +++
 src/contract_check/worker_notify/__init__.py  |   0
 src/contract_check/worker_notify/__main__.py  |  62 +++
 src/contract_check/worker_notify/consumer.py  |  61 +++
 src/contract_check/worker_notify/handler.py   |  78 ++++
 srv/worker-notify/Dockerfile                  |  34 ++
 tests/unit/test_web_auth.py                   | 223 ++++++++++
 uv.lock                                       |  63 +++
 31 files changed, 1772 insertions(+), 40 deletions(-)
 create mode 100644 migrations/versions/0003_web_auth.py
 create mode 100644 src/contract_check/core/auth_refresh.py
 create mode 100644 src/contract_check/core/auth_refresh_key.py
 create mode 100644 src/contract_check/core/notifications/__init__.py
 create mode 100644 src/contract_check/core/notifications/publisher.py
 create mode 100644 src/contract_check/core/notifications/transport.py
 create mode 100644 src/contract_check/core/security/__init__.py
 create mode 100644 src/contract_check/core/security/passwords.py
 create mode 100644 src/contract_check/worker_notify/__init__.py
 create mode 100644 src/contract_check/worker_notify/__main__.py
 create mode 100644 src/contract_check/worker_notify/consumer.py
 create mode 100644 src/contract_check/worker_notify/handler.py
 create mode 100644 srv/worker-notify/Dockerfile
 create mode 100644 tests/unit/test_web_auth.py

diff --git a/.env.example b/.env.example
index 21c80ee..5742eb3 100644
--- a/.env.example
+++ b/.env.example
@@ -19,6 +19,7 @@ REDIS_URL=redis://redis:6379/0
 RABBITMQ_URL=amqp://contract_check:contract_check@rabbitmq:5672//
 MQ_PREFETCH_EXTRACT=1         # CPU-bound extraction; tune to CPU count
 MQ_PREFETCH_ANALYZE=3         # mirrors Ollama Pro concurrency
+MQ_PREFETCH_NOTIFY=5          # notification worker (I/O bound)
 MQ_MAX_ATTEMPTS=5             # before a message lands on the DLQ
 MQ_RETRY_BASE_MS=2000         # exponential backoff base (2s, 4s, 8s, ...)
 
@@ -67,6 +68,21 @@ TELEGRAM_BOT_TOKEN=
 JWT_SECRET=                  # HS256 secret for signing user JWTs; generate with `openssl rand -hex 32`
 JWT_ALGORITHM=HS256
 JWT_ACCESS_TTL_MINUTES=1440  # 24 hours default
+JWT_REFRESH_TTL_DAYS=30      # refresh-token lifetime for webUI auth
+
+# --- WebUI auth (email + password) ---
+WEB_AUTH_ENABLED=true        # toggle /api/v1/auth/{register,login,...} routes
+PASSWORD_RESET_TTL_MINUTES=60
+PASSWORD_MIN_LENGTH=8
+WEB_APP_BASE_URL=http://localhost:5173   # SPA base — used to build reset links
+
+# --- SMTP (notification transport; empty host → dev logger) ---
+SMTP_HOST=
+SMTP_PORT=587
+SMTP_USERNAME=
+SMTP_PASSWORD=
+SMTP_FROM=no-reply@contract-check.local
+SMTP_USE_TLS=true
 
 # --- Telegram bot (adapter, HTTP-only to api) ---
 BOT_TOKEN=                    # same value as TELEGRAM_BOT_TOKEN (kept for the bot image)
diff --git a/Makefile b/Makefile
index cd0dc1b..bab56c2 100644
--- a/Makefile
+++ b/Makefile
@@ -3,8 +3,9 @@
 
 .PHONY: help install lint typecheck test test-unit test-integration migrate \
         infra-up infra-down infra-logs services-up services-down services-logs \
-        api api-logs bot bot-logs worker-extract worker-analyze seed-token \
-        jwt-secret jwt-token jwt-verify health shell-api shell-bot shell-db clean
+        api api-logs bot bot-logs worker-extract worker-analyze worker-notify \
+        seed-token jwt-secret jwt-token jwt-verify health shell-api shell-bot \
+        shell-db clean
 
 # ─────────────────────────────────────────────────────────────────────────────
 # Help
@@ -86,6 +87,12 @@ worker-analyze:  ## Start/restart analyze worker
 worker-analyze-logs:  ## Tail analyze worker logs
 	docker compose logs -f worker-analyze
 
+worker-notify:  ## Start/restart notify worker
+	docker compose --profile services up -d --build --remove-orphans worker-notify
+
+worker-notify-logs:  ## Tail notify worker logs
+	docker compose logs -f worker-notify
+
 # ─────────────────────────────────────────────────────────────────────────────
 # Database
 # ─────────────────────────────────────────────────────────────────────────────
diff --git a/docker-compose.yml b/docker-compose.yml
index 4f9386f..609d299 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -170,6 +170,17 @@ services:
       JWT_SECRET: ${JWT_SECRET}
       JWT_ALGORITHM: ${JWT_ALGORITHM:-HS256}
       JWT_ACCESS_TTL_MINUTES: ${JWT_ACCESS_TTL_MINUTES:-1440}
+      JWT_REFRESH_TTL_DAYS: ${JWT_REFRESH_TTL_DAYS:-30}
+      WEB_AUTH_ENABLED: ${WEB_AUTH_ENABLED:-true}
+      PASSWORD_RESET_TTL_MINUTES: ${PASSWORD_RESET_TTL_MINUTES:-60}
+      PASSWORD_MIN_LENGTH: ${PASSWORD_MIN_LENGTH:-8}
+      WEB_APP_BASE_URL: ${WEB_APP_BASE_URL:-http://localhost:5173}
+      SMTP_HOST: ${SMTP_HOST:-}
+      SMTP_PORT: ${SMTP_PORT:-587}
+      SMTP_USERNAME: ${SMTP_USERNAME:-}
+      SMTP_PASSWORD: ${SMTP_PASSWORD:-}
+      SMTP_FROM: ${SMTP_FROM:-no-reply@contract-check.local}
+      SMTP_USE_TLS: ${SMTP_USE_TLS:-true}
     ports:
       - "${API_PORT:-8000}:8000"
       - "${API_METRICS_PORT:-9100}:9100"
@@ -270,6 +281,37 @@ services:
     ports:
       - "9102:9102"
 
+  worker-notify:
+    profiles: ["services"]
+    build:
+      context: .
+      dockerfile: srv/worker-notify/Dockerfile
+    container_name: contract_check-worker-notify
+    restart: unless-stopped
+    depends_on:
+      rabbitmq:
+        condition: service_healthy
+    environment:
+      ENV: ${ENV:-dev}
+      LOG_LEVEL: ${LOG_LEVEL:-INFO}
+      DATABASE_URL: postgresql+asyncpg://${POSTGRES_USER:-contract_check}:${POSTGRES_PASSWORD:-contract_check}@postgres:5432/${POSTGRES_DB:-contract_check}
+      RABBITMQ_URL: amqp://${RABBITMQ_USER:-contract_check}:${RABBITMQ_PASS:-contract_check}@rabbitmq:5672/${RABBITMQ_VHOST:-/}
+      SENTRY_DSN: ${SENTRY_DSN:-}
+      OTEL_EXPORTER_OTLP_ENDPOINT: ${OTEL_EXPORTER_OTLP_ENDPOINT:-}
+      OTEL_SERVICE_NAME: worker-notify
+      MQ_PREFETCH_NOTIFY: ${MQ_PREFETCH_NOTIFY:-5}
+      MQ_MAX_ATTEMPTS: ${MQ_MAX_ATTEMPTS:-5}
+      MQ_RETRY_BASE_MS: ${MQ_RETRY_BASE_MS:-2000}
+      SMTP_HOST: ${SMTP_HOST:-}
+      SMTP_PORT: ${SMTP_PORT:-587}
+      SMTP_USERNAME: ${SMTP_USERNAME:-}
+      SMTP_PASSWORD: ${SMTP_PASSWORD:-}
+      SMTP_FROM: ${SMTP_FROM:-no-reply@contract-check.local}
+      SMTP_USE_TLS: ${SMTP_USE_TLS:-true}
+      WEB_APP_BASE_URL: ${WEB_APP_BASE_URL:-http://localhost:5173}
+    ports:
+      - "9103:9103"
+
   # Telegram bot adapter (aiogram 3, HTTP-only to api). Per docs/ARCHITECTURE.md §17
   # the bot holds no DB/MQ/S3 credentials — it depends on `api` being healthy,
   # not on the infra containers directly, enforcing the hexagonal boundary even
diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md
index 8ad6e4e..20bb57a 100644
--- a/docs/ARCHITECTURE.md
+++ b/docs/ARCHITECTURE.md
@@ -355,11 +355,12 @@ class DocumentUploaded(BaseModel):
     correlation_id: UUID
     document_id: UUID
     user_id: UUID
-    s3_key: str            # users/{uid}/docs/{did}.{ext}
+    s3_key: str  # users/{uid}/docs/{did}.{ext}
     filename: str
     mime: str
     attempt: int = 0
 
+
 class DocumentExtracted(BaseModel):
     correlation_id: UUID
     document_id: UUID
@@ -554,6 +555,7 @@ double-refund:
 ```python
 NON_REFUNDABLE_INFRA_ONLY = {"extraction_failed"}  # user-garbage input
 
+
 async def refund_credit(session, document_id, failure_class, policy) -> bool:
     if policy == "infra_only" and failure_class in NON_REFUNDABLE_INFRA_ONLY:
         return False  # user pays for undetectable garbage
@@ -606,9 +608,10 @@ GigaChat, or YandexGPT without touching `analyzer.py`.
 from typing import Protocol, Sequence
 from dataclasses import dataclass
 
+
 @dataclass(slots=True)
 class AnalysisResult:
-    findings: list           # list[Finding] (from core.analysis.report_schema)
+    findings: list  # list[Finding] (from core.analysis.report_schema)
     model_used: str
     fell_back: bool
     repaired: bool
@@ -616,6 +619,7 @@ class AnalysisResult:
     eval_tokens: int
     latency_sec: float
 
+
 class LLMProvider(Protocol):
     async def analyze(self, text: str, *, checklist: str) -> AnalysisResult: ...
     async def aclose(self) -> None: ...
@@ -647,8 +651,10 @@ Map adapter-internal failures to `FailureClass`:
 ```python
 def build_llm_provider(settings) -> LLMProvider:
     match settings.llm_provider:
-        case "ollama_cloud": return OllamaCloudProvider(settings)
-        case _: raise ValueError(f"unknown LLM_PROVIDER={settings.llm_provider!r}")
+        case "ollama_cloud":
+            return OllamaCloudProvider(settings)
+        case _:
+            raise ValueError(f"unknown LLM_PROVIDER={settings.llm_provider!r}")
 ```
 
 `LLM_PROVIDER` env (default `ollama_cloud`). Future providers register here.
diff --git a/migrations/versions/0003_web_auth.py b/migrations/versions/0003_web_auth.py
new file mode 100644
index 0000000..95c29dc
--- /dev/null
+++ b/migrations/versions/0003_web_auth.py
@@ -0,0 +1,56 @@
+"""Email/password columns on users + webUI auth (Stage 4).
+
+Revision ID: 0003
+Revises: 0002
+Create Date: 2026-08-12
+
+Adds nullable email/password_hash/reset columns to the existing Telegram-only
+users table. Existing Telegram rows remain valid (telegram_id still satisfies
+the new users_identity_present check). Email is unique; both telegram_id and
+email may co-exist on one row (linking the two identities later is a no-op).
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "0003"
+down_revision: str | None = "0002"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.add_column("users", sa.Column("email", sa.Text(), nullable=True))
+    op.add_column("users", sa.Column("password_hash", sa.Text(), nullable=True))
+    op.add_column("users", sa.Column("password_reset_token_hash", sa.Text(), nullable=True))
+    op.add_column(
+        "users", sa.Column("password_reset_expires_at", sa.DateTime(timezone=True), nullable=True)
+    )
+    op.add_column(
+        "users",
+        sa.Column("is_active", sa.Boolean(), nullable=False, server_default=sa.text("true")),
+    )
+
+    op.create_unique_constraint("users_email_unique", "users", ["email"])
+
+    # A user must have at least one identity anchor (telegram_id OR email).
+    op.create_check_constraint(
+        "users_identity_present",
+        "users",
+        "telegram_id IS NOT NULL OR email IS NOT NULL",
+    )
+
+
+def downgrade() -> None:
+    op.drop_constraint("users_identity_present", "users", type_="check")
+    op.drop_constraint("users_email_unique", "users", type_="unique")
+    op.drop_column("users", "is_active")
+    op.drop_column("users", "password_reset_expires_at")
+    op.drop_column("users", "password_reset_token_hash")
+    op.drop_column("users", "password_hash")
+    op.drop_column("users", "email")
diff --git a/pyproject.toml b/pyproject.toml
index a62cbc4..87a23e7 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -70,6 +70,8 @@ api = [
     "python-multipart>=0.0.9",
     "redis>=5.0",
     "pyjwt[crypto]>=2.8",
+    "argon2-cffi>=23.1",
+    "email-validator>=2.1",
     "opentelemetry-instrumentation-fastapi>=0.45b0",
     "opentelemetry-instrumentation-asgi>=0.45b0",
     "opentelemetry-instrumentation-httpx>=0.45b0",
@@ -93,6 +95,12 @@ analyze = [
 bot = [
     "aiogram>=3.4",
 ]
+notify = [
+    { include-group = "db" },
+    { include-group = "mq" },
+    { include-group = "obs" },
+    "aiosmtplib>=3.0",
+]
 prototype = [
     "pymupdf>=1.24",
     "python-docx>=1.1",
@@ -102,6 +110,7 @@ dev = [
     { include-group = "extract" },
     { include-group = "analyze" },
     { include-group = "bot" },
+    { include-group = "notify" },
     { include-group = "prototype" },
     "pytest>=8",
     "pytest-asyncio>=0.23",
diff --git a/src/contract_check/api/app.py b/src/contract_check/api/app.py
index 48d1bba..01abac4 100644
--- a/src/contract_check/api/app.py
+++ b/src/contract_check/api/app.py
@@ -15,6 +15,7 @@ from ..core.llm import port as llm_port  # noqa: F401 — package loaded
 from ..core.logging import bind_context, configure_logging, get_logger
 from ..core.metrics import redis_connected
 from ..core.mq.publisher import Publisher
+from ..core.notifications.publisher import NotificationPublisher
 from ..core.rate_limit import MemoryRateLimiter, RateLimiter, RedisRateLimiter
 from ..core.redis_client import get_redis_client
 from ..core.s3.minio_storage import MinioStorage
@@ -49,6 +50,9 @@ async def lifespan(app: FastAPI) -> Any:
     publisher = Publisher(settings.rabbitmq_url, origin="api")
     await publisher.connect()
 
+    notification_publisher = NotificationPublisher(settings.rabbitmq_url, origin="api")
+    await notification_publisher.connect()
+
     # Redis is used for rate-limiting/sessions. In dev/tests without Redis we
     # transparently fall back to an in-memory bucket so unit tests stay
     # dependency-free.
@@ -63,18 +67,25 @@ async def lifespan(app: FastAPI) -> Any:
         log.warning("redis_unavailable", redis_url=settings.redis_url, error=str(exc))
         rate_limiter = MemoryRateLimiter()
         redis_connected.set(0)
+        # webUI auth cannot work without Redis (refresh-token store). Wipe the
+        # handle so the RefreshStoreDep raises 503 rather than silently no-op'ing.
+        redis_client = None
 
     app.state.storage = storage
     app.state.publisher = publisher
+    app.state.notification_publisher = notification_publisher
     app.state.rate_limiter = rate_limiter
+    app.state.redis = redis_client
 
     yield
 
     await publisher.close()
-    try:
-        await redis_client.aclose()
-    except Exception:
-        pass
+    await notification_publisher.close()
+    if redis_client is not None:
+        try:
+            await redis_client.aclose()
+        except Exception:
+            pass
     shutdown_telemetry()
 
 
diff --git a/src/contract_check/api/deps.py b/src/contract_check/api/deps.py
index a48efcb..66c4250 100644
--- a/src/contract_check/api/deps.py
+++ b/src/contract_check/api/deps.py
@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 from collections.abc import AsyncIterator
-from typing import Annotated
+from typing import Annotated, Any
 from uuid import UUID
 
 from fastapi import Depends, Header, HTTPException, Request
@@ -12,11 +12,13 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from ..core.api_keys import hash_api_key
 from ..core.auth import AuthError, TokenExpiredError, TokenInvalidError, verify_access_token
+from ..core.auth_refresh import RefreshTokenStore
 from ..core.config import get_settings
 from ..core.db.models import User
 from ..core.db.session import create_session_factory
 from ..core.logging import get_logger
 from ..core.mq.publisher import Publisher
+from ..core.notifications.publisher import NotificationPublisher
 from ..core.rate_limit import RateLimiter
 from ..core.s3.port import Storage
 from ..core.tokens import hash_token
@@ -49,6 +51,41 @@ def get_publisher(request: Request) -> Publisher:
 PublisherDep = Annotated[Publisher, Depends(get_publisher)]
 
 
+def get_notification_publisher(request: Request) -> NotificationPublisher:
+    publisher: NotificationPublisher = request.app.state.notification_publisher
+    return publisher
+
+
+NotificationPublisherDep = Annotated[NotificationPublisher, Depends(get_notification_publisher)]
+
+
+def get_redis(request: Request) -> Any:
+    """Return the app-state async Redis client.
+
+    Set on app.state.redis in the api lifespan. Used by the refresh-token store.
+    """
+    redis: Any = request.app.state.redis
+    return redis
+
+
+RedisDep = Annotated[Any, Depends(get_redis)]
+
+
+def get_refresh_store(request: Request) -> RefreshTokenStore:
+    """Build a RefreshTokenStore from the app-state Redis client.
+
+    Raises 503 if Redis is unavailable — webUI auth cannot function without it.
+    """
+    redis: Any = getattr(request.app.state, "redis", None)
+    if redis is None:
+        raise HTTPException(status_code=503, detail="refresh-token store unavailable")
+    settings = get_settings()
+    return RefreshTokenStore(redis, ttl_seconds=settings.jwt_refresh_ttl_days * 24 * 3600)
+
+
+RefreshStoreDep = Annotated[RefreshTokenStore, Depends(get_refresh_store)]
+
+
 async def require_service_token(
     session: AsyncSessionDep,
     authorization: Annotated[str | None, Header()] = None,
@@ -115,6 +152,76 @@ async def get_or_create_user_by_id(session: AsyncSession, user_id: UUID) -> User
     return User(id=row[0], telegram_id=row[1], created_at=row[2], credits_left=row[3])
 
 
+async def fetch_user_by_email(session: AsyncSession, email: str) -> User | None:
+    """Fetch a user by email (case-sensitive — normalize upstream). Returns None if not found."""
+    result = await session.execute(
+        text(
+            "SELECT id, telegram_id, email, password_hash, is_active, created_at, credits_left "
+            "FROM users WHERE email = :e"
+        ),
+        {"e": email},
+    )
+    row = result.first()
+    if row is None:
+        return None
+    return User(
+        id=row[0],
+        telegram_id=row[1],
+        email=row[2],
+        password_hash=row[3],
+        is_active=row[4],
+        created_at=row[5],
+        credits_left=row[6],
+    )
+
+
+async def fetch_user_by_id_full(session: AsyncSession, user_id: UUID) -> User | None:
+    """Fetch a user by UUID including web-auth columns."""
+    result = await session.execute(
+        text(
+            "SELECT id, telegram_id, email, password_hash, is_active, created_at, credits_left "
+            "FROM users WHERE id = :u"
+        ),
+        {"u": user_id},
+    )
+    row = result.first()
+    if row is None:
+        return None
+    return User(
+        id=row[0],
+        telegram_id=row[1],
+        email=row[2],
+        password_hash=row[3],
+        is_active=row[4],
+        created_at=row[5],
+        credits_left=row[6],
+    )
+
+
+async def create_email_user(session: AsyncSession, *, email: str, password_hash: str) -> User:
+    """Insert a new email/password user with 0 credits and return it."""
+    result = await session.execute(
+        text(
+            "INSERT INTO users (email, password_hash, credits_left) "
+            "VALUES (:e, :p, 0) "
+            "RETURNING id, telegram_id, email, password_hash, is_active, created_at, credits_left"
+        ),
+        {"e": email, "p": password_hash},
+    )
+    row = result.first()
+    assert row is not None
+    await session.commit()
+    return User(
+        id=row[0],
+        telegram_id=row[1],
+        email=row[2],
+        password_hash=row[3],
+        is_active=row[4],
+        created_at=row[5],
+        credits_left=row[6],
+    )
+
+
 def get_rate_limiter(request: Request) -> RateLimiter:
     """Return the app-state rate limiter (Redis in prod, Memory in tests)."""
     limiter: RateLimiter = request.app.state.rate_limiter
diff --git a/src/contract_check/api/routes/README.md b/src/contract_check/api/routes/README.md
index fc1e46f..e1de3f0 100644
--- a/src/contract_check/api/routes/README.md
+++ b/src/contract_check/api/routes/README.md
@@ -166,6 +166,103 @@ Mini App initData. **Auth:** нет (проверяется подпись ` ini
 
 ---
 
+### WebUI auth (email + password)
+
+Шесть эндпоинтов для SPA-фронта. Возвращают **JWT-пару**: короткий access
+(default 24h) + длинный refresh (default 30d, ttl в Redis). Refresh-token
+поддерживает отзыв через logout; access невалидируется только истечением TTL.
+
+| Метод | Путь                        | Auth   | Описание                                |
+| ----- | --------------------------- | ------ | --------------------------------------- |
+| POST  | `/api/v1/auth/register`     | —      | Регистрация → 201 + JWT pair            |
+| POST  | `/api/v1/auth/login`        | —      | Вход → 200 + JWT pair                   |
+| POST  | `/api/v1/auth/logout`       | —      | Отзыв refresh-токена (в теле)          |
+| GET   | `/api/v1/auth/me`           | Bearer | Текущий пользователь (расширенный ответ)|
+| POST  | `/api/v1/auth/forgot-password` | —   | Запрос сброса (202, всегда одинаковый ответ) |
+| POST  | `/api/v1/auth/reset-password`  | —   | Сброс пароля по токену из письма       |
+
+#### `POST /api/v1/auth/register`
+
+Тело — `RegisterRequest`:
+
+| Поле       | Тип     | Условие         |
+| ---------- | ------- | --------------- |
+| `email`    | EmailStr | валидный email |
+| `password` | str     | `8..128` символов |
+
+`409` если email занят. `201` — `TokenPairResponse`.
+
+#### `POST /api/v1/auth/login`
+
+Тело — `LoginRequest` (те же поля). `401` при неверных кредах.
+
+#### `TokenPairResponse`
+
+```json
+{
+  "access_token": "<jwt>",
+  "refresh_token": "<jwt>",
+  "token_type": "bearer",
+  "expires_in": 86400,
+  "user": {
+    "id": "uuid",
+    "email": "user@example.com",
+    "telegram_id": null,
+    "credits_left": 0,
+    "is_active": true,
+    "created_at": "2026-08-12T..."
+  }
+}
+```
+
+#### `POST /api/v1/auth/logout`
+
+Тело — `LogoutRequest`:
+
+| Поле            | Тип |
+| --------------- | --- |
+| `refresh_token` | str |
+
+Идемпотентен. Access-токен остаётся валидным до истечения своего TTL
+(см. `JWT_ACCESS_TTL_MINUTES`); refresh уничтожается в Redis сразу.
+
+#### `POST /api/v1/auth/forgot-password`
+
+Тело — `ForgotPasswordRequest` (`email`). Всегда отвечает `202 OkResponse` с
+телом `{"ok": true, "detail": "if the email exists, a reset link was sent"}`
+— чтобы не раскрывать, какие адреса зарегистрированы. API:
+
+1. Генерирует `secrets.token_urlsafe(32)`, хранит только SHA-256 от него
+   в `users.password_reset_token_hash` с TTL = `PASSWORD_RESET_TTL_MINUTES`.
+2. Публикует `NotificationMessage(kind=password_reset)` в очередь
+   `notify.q` (RabbitMQ).
+3. `worker-notify` достаёт сообщение и шлёт письмо через SMTP
+   (`SMTP_HOST`/`SMTP_PORT`/`SMTP_USE_TLS`/`SMTP_FROM`). При пустом
+   `SMTP_HOST` (dev) тело письма пишется в лог.
+
+#### `POST /api/v1/auth/reset-password`
+
+Тело — `ResetPasswordRequest`:
+
+| Поле       | Тип | Условие         |
+| ---------- | --- | --------------- |
+| `token`    | str | из письма       |
+| `password` | str | `8..128` символов |
+
+`400` при невалидном/просроченном токене. Успех: пароль перезаписывается
+(argon2id), `password_reset_token_hash` сбрасывается, **все активные
+refresh-токены этого пользователя отзываются** в Redis (принудительный
+re-login на всех устройствах).
+
+#### `GET /api/v1/auth/me`
+
+Расширение прежнего introspect-эндпоинта: теперь читает Bearer access JWT,
+ищет пользователя в БД и возвращает полный профиль. Старые поля
+(`sub`, `telegram_id`, `type`, `exp`) сохранены для совместимости; добавлены
+`email`, `credits_left`, `is_active`, `created_at`.
+
+---
+
 ## me
 
 Файл: [`me.py`](me.py). Тег: `me`. **Auth:** `CurrentUserDep` (user JWT).
diff --git a/src/contract_check/api/routes/auth.py b/src/contract_check/api/routes/auth.py
index 6b430cf..b6c563b 100644
--- a/src/contract_check/api/routes/auth.py
+++ b/src/contract_check/api/routes/auth.py
@@ -1,33 +1,63 @@
-"""Telegram-based authentication endpoints.
+"""Authentication endpoints: Telegram identity sources + webUI email/password.
 
-Three identity sources converge on the same JWT:
+Three Telegram identity sources converge on the same access JWT:
   - /auth/telegram/bot    — bot adapter exchanges a verified telegram_id for JWT
   - /auth/telegram/web    — Telegram Login Widget callback
   - /auth/telegram/miniapp  — Mini App initData
 
-Protected user endpoints receive the JWT via `Authorization: Bearer <jwt>` and use
-`deps.CurrentUser`.
+WebUI (email/password) endpoints issue a JWT pair (access + refresh):
+  - POST /auth/register         — email + password -> user + pair
+  - POST /auth/login            — email + password -> pair
+  - POST /auth/logout           — revoke refresh
+  - GET  /auth/me               — current user (Bearer access JWT)
+  - POST /auth/forgot-password  — store reset-token hash, enqueue notification
+  - POST /auth/reset-password   — verify token, set new password
+
+Refresh tokens carry a `jti` tracked in Redis so logout is enforceable.
+Password reset tokens are stored as SHA-256 hashes with an expiry in the
+users row; the actual delivery link is built by the notify worker.
+Protected user endpoints receive the access JWT via
+`Authorization: Bearer <jwt>` and use `deps.CurrentUser`.
 """
 
 from __future__ import annotations
 
+import datetime as dt
+import hashlib
+import secrets
+import uuid
 from typing import Annotated, Any
 
 from fastapi import APIRouter, Header, HTTPException, status
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, EmailStr, Field
 
 from ...core.auth import (
     AuthError,
     create_access_token,
+    create_refresh_token,
     verify_access_token,
     verify_bot_identity,
+    verify_refresh_token,
     verify_telegram_miniapp_init_data,
     verify_telegram_web_payload,
 )
 from ...core.config import get_settings
 from ...core.db.models import User
-from ..deps import AsyncSessionDep, AuthDep, get_or_create_user_for_telegram
+from ...core.logging import get_logger
+from ...core.mq.messages import NotificationMessage
+from ...core.security.passwords import hash_password, verify_password
+from ..deps import (
+    AsyncSessionDep,
+    AuthDep,
+    NotificationPublisherDep,
+    RefreshStoreDep,
+    create_email_user,
+    fetch_user_by_email,
+    fetch_user_by_id_full,
+    get_or_create_user_for_telegram,
+)
 
+log = get_logger(__name__)
 router = APIRouter(tags=["auth"])
 
 
@@ -64,6 +94,26 @@ class TokenIntrospectResponse(BaseModel):
     exp: int
 
 
+class WebUserPublic(BaseModel):
+    """User profile subset safe to return to the webUI."""
+
+    id: str
+    email: str | None = None
+    telegram_id: int | None = None
+    credits_left: int
+    is_active: bool
+    created_at: dt.datetime
+
+
+class MeResponse(TokenIntrospectResponse):
+    """Extends the introspection shape with web-user fields (additive)."""
+
+    email: str | None = None
+    credits_left: int = 0
+    is_active: bool = True
+    created_at: dt.datetime | None = None
+
+
 def _issue_token(user: User) -> AuthResponse:
     token = create_access_token(user.id, user.telegram_id or 0)
     ttl = get_settings().jwt_access_ttl_minutes * 60
@@ -120,25 +170,30 @@ async def auth_telegram_miniapp(
     return _issue_token(user)
 
 
-@router.get("/api/v1/auth/me", status_code=status.HTTP_200_OK)
-async def introspect_token(
+@router.get("/api/v1/auth/me", response_model=MeResponse, status_code=status.HTTP_200_OK)
+async def me(
+    session: AsyncSessionDep,
     authorization: Annotated[str | None, Header()] = None,
-) -> TokenIntrospectResponse:
-    """Debug/introspection endpoint: return claims for a Bearer JWT."""
-    if not authorization or not authorization.lower().startswith("bearer "):
-        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="missing bearer token")
+) -> MeResponse:
+    """Current user — verifies the Bearer JWT and returns the user's profile.
 
-    token = authorization[7:].strip()
-    try:
-        claims = verify_access_token(token)
-    except AuthError as exc:
-        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=str(exc)) from exc
+    Backward-compatible with the previous introspect shape (sub/telegram_id/
+    type/exp) and adds email/credits_left/is_active/created_at.
+    """
+    claims = _require_access_claims(authorization)
+    user = await fetch_user_by_id_full(session, claims.sub)
+    if user is None:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="user not found")
 
-    return TokenIntrospectResponse(
-        sub=str(claims.sub),
-        telegram_id=claims.telegram_id,
+    return MeResponse(
+        sub=str(user.id),
+        telegram_id=user.telegram_id or 0,
         type=claims.type,
         exp=claims.exp or 0,
+        email=user.email,
+        credits_left=user.credits_left,
+        is_active=bool(user.is_active),
+        created_at=user.created_at,
     )
 
 
@@ -146,3 +201,307 @@ async def introspect_token(
 async def token_permissions_dummy() -> dict[str, Any]:
     """Placeholder for future RBAC expansion."""
     return {"permissions": ["upload", "read_reports", "read_me"]}
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# WebUI auth (email + password)
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+class RegisterRequest(BaseModel):
+    email: EmailStr
+    password: str = Field(..., min_length=8, max_length=128)
+
+
+class LoginRequest(BaseModel):
+    email: EmailStr
+    password: str = Field(..., min_length=1, max_length=128)
+
+
+class TokenPairResponse(BaseModel):
+    """JWT pair returned by register/login."""
+
+    access_token: str
+    refresh_token: str
+    token_type: str = "bearer"
+    expires_in: int  # access TTL in seconds
+    user: WebUserPublic
+
+
+class LogoutRequest(BaseModel):
+    refresh_token: str = Field(..., min_length=1)
+
+
+class ForgotPasswordRequest(BaseModel):
+    email: EmailStr
+
+
+class ResetPasswordRequest(BaseModel):
+    token: str = Field(..., min_length=1, max_length=256)
+    password: str = Field(..., min_length=8, max_length=128)
+
+
+class OkResponse(BaseModel):
+    """Generic `{ok: true, ...}` payload for state-mutating auth endpoints."""
+
+    ok: bool = True
+    detail: str | None = None
+
+
+def _require_access_claims(authorization: str | None) -> Any:
+    """Shared bearer-extraction used by `me` and any future stateless endpoint."""
+    if not authorization or not authorization.lower().startswith("bearer "):
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="missing bearer token")
+    token = authorization[7:].strip()
+    try:
+        return verify_access_token(token)
+    except AuthError as exc:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=str(exc)) from exc
+
+
+def _web_user_public(user: User) -> WebUserPublic:
+    return WebUserPublic(
+        id=str(user.id),
+        email=user.email,
+        telegram_id=user.telegram_id,
+        credits_left=user.credits_left,
+        is_active=bool(user.is_active),
+        created_at=user.created_at,
+    )
+
+
+async def _issue_pair(user: User, refresh_store: RefreshStoreDep) -> TokenPairResponse:
+    """Mint an access + refresh pair for a verified user."""
+    settings = get_settings()
+    access = create_access_token(user.id, user.telegram_id or 0)
+    jti = await refresh_store.issue(user.id)
+    refresh = create_refresh_token(user.id, jti)
+    return TokenPairResponse(
+        access_token=access,
+        refresh_token=refresh,
+        token_type="bearer",
+        expires_in=settings.jwt_access_ttl_minutes * 60,
+        user=_web_user_public(user),
+    )
+
+
+def _hash_reset_token(token: str) -> str:
+    """SHA-256 of a reset token — store this, never the raw token."""
+    return hashlib.sha256(token.encode("utf-8")).hexdigest()
+
+
+def _build_reset_link(token: str) -> str:
+    base = get_settings().web_app_base_url.rstrip("/")
+    return f"{base}/reset-password?token={token}"
+
+
+@router.post(
+    "/api/v1/auth/register",
+    response_model=TokenPairResponse,
+    status_code=status.HTTP_201_CREATED,
+)
+async def register(
+    session: AsyncSessionDep,
+    refresh_store: RefreshStoreDep,
+    body: RegisterRequest,
+) -> TokenPairResponse:
+    """Register a new email/password user and issue a JWT pair."""
+    _require_web_auth_enabled()
+    email_normalized = body.email.lower().strip()
+
+    existing = await fetch_user_by_email(session, email_normalized)
+    if existing is not None:
+        # Do not leak which emails are registered.
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="email already registered")
+
+    hashed = hash_password(body.password)
+    user = await create_email_user(session, email=email_normalized, password_hash=hashed)
+    log.info("user_registered", user_id=str(user.id), email=email_normalized)
+    return await _issue_pair(user, refresh_store)
+
+
+@router.post(
+    "/api/v1/auth/login",
+    response_model=TokenPairResponse,
+    status_code=status.HTTP_200_OK,
+)
+async def login(
+    session: AsyncSessionDep,
+    refresh_store: RefreshStoreDep,
+    body: LoginRequest,
+) -> TokenPairResponse:
+    """Email + password -> JWT pair."""
+    _require_web_auth_enabled()
+    email_normalized = body.email.lower().strip()
+
+    user = await fetch_user_by_email(session, email_normalized)
+    if user is None or not user.password_hash:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid credentials")
+    if not verify_password(body.password, user.password_hash):
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid credentials")
+    if not user.is_active:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="account disabled")
+
+    log.info("user_logged_in", user_id=str(user.id))
+    return await _issue_pair(user, refresh_store)
+
+
+@router.post(
+    "/api/v1/auth/logout",
+    response_model=OkResponse,
+    status_code=status.HTTP_200_OK,
+)
+async def logout(
+    refresh_store: RefreshStoreDep,
+    body: LogoutRequest,
+) -> OkResponse:
+    """Revoke the supplied refresh token. Access token stays valid until it expires."""
+    _require_web_auth_enabled()
+    try:
+        claims = verify_refresh_token(body.refresh_token)
+    except AuthError as exc:
+        # Already expired or invalid — nothing to revoke. Return ok for idempotency.
+        log.info("logout_invalid_refresh", error=str(exc))
+        return OkResponse(ok=True, detail="already revoked")
+    removed = await refresh_store.revoke(claims.sub, claims.jti)
+    log.info("user_logged_out", user_id=str(claims.sub), removed=removed)
+    return OkResponse(ok=True, detail="refresh revoked")
+
+
+@router.post(
+    "/api/v1/auth/forgot-password",
+    response_model=OkResponse,
+    status_code=status.HTTP_202_ACCEPTED,
+)
+async def forgot_password(
+    session: AsyncSessionDep,
+    publisher: NotificationPublisherDep,
+    body: ForgotPasswordRequest,
+) -> OkResponse:
+    """Generate a reset token, store its hash + expiry, and enqueue a notification.
+
+    Always returns 202 with `ok=true` regardless of whether the email exists,
+    to avoid leaking which addresses are registered.
+    """
+    _require_web_auth_enabled()
+    email_normalized = body.email.lower().strip()
+
+    user = await fetch_user_by_email(session, email_normalized)
+    if user is None:
+        log.info("forgot_password_unknown_email", email=email_normalized)
+        return OkResponse(ok=True, detail="if the email exists, a reset link was sent")
+
+    settings = get_settings()
+    raw_token = secrets.token_urlsafe(32)
+    token_hash = _hash_reset_token(raw_token)
+    expires_at = dt.datetime.now(tz=dt.UTC) + dt.timedelta(
+        minutes=settings.password_reset_ttl_minutes
+    )
+
+    from sqlalchemy import text as sa_text
+
+    await session.execute(
+        sa_text(
+            "UPDATE users "
+            "SET password_reset_token_hash = :h, password_reset_expires_at = :e "
+            "WHERE id = :u"
+        ),
+        {"h": token_hash, "e": expires_at, "u": user.id},
+    )
+    await session.commit()
+
+    reset_link = _build_reset_link(raw_token)
+    notification = NotificationMessage(
+        correlation_id=uuid.uuid4(),
+        kind="password_reset",
+        to=email_normalized,
+        subject="Восстановление пароля — Контракт-чек",
+        body_text=(
+            "Вы запросили сброс пароля.\n\n"
+            f"Перейдите по ссылке, чтобы задать новый пароль (действует "
+            f"{settings.password_reset_ttl_minutes} мин.):\n{reset_link}\n\n"
+            "Если вы не запрашивали сброс — просто проигнорируйте это письмо."
+        ),
+        body_html=(
+            "<p>Вы запросили сброс пароля.</p>"
+            f'<p><a href="{reset_link}">Задать новый пароль</a> '
+            f"(действует {settings.password_reset_ttl_minutes} мин.)</p>"
+            "<p>Если вы не запрашивали сброс — проигнорируйте это письмо.</p>"
+        ),
+    )
+    try:
+        await publisher.publish(notification, routing_key="notify")
+    except Exception as exc:  # noqa: BLE001 — best-effort; reset is still storable
+        log.error(
+            "forgot_password_publish_failed",
+            user_id=str(user.id),
+            error=str(exc),
+        )
+
+    log.info("forgot_password_enqueued", user_id=str(user.id))
+    return OkResponse(ok=True, detail="if the email exists, a reset link was sent")
+
+
+@router.post(
+    "/api/v1/auth/reset-password",
+    response_model=OkResponse,
+    status_code=status.HTTP_200_OK,
+)
+async def reset_password(
+    session: AsyncSessionDep,
+    refresh_store: RefreshStoreDep,
+    body: ResetPasswordRequest,
+) -> OkResponse:
+    """Verify a reset token and set the new password.
+
+    On success: clears the stored token hash, rotates the password, and revokes
+    all active refresh tokens for the user (forcing re-login everywhere).
+    """
+    _require_web_auth_enabled()
+    token_hash = _hash_reset_token(body.token)
+
+    from sqlalchemy import text as sa_text
+
+    result = await session.execute(
+        sa_text(
+            "SELECT id, password_reset_expires_at FROM users WHERE password_reset_token_hash = :h"
+        ),
+        {"h": token_hash},
+    )
+    row = result.first()
+    if row is None:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="invalid reset token")
+
+    user_id = row[0]
+    expires_at = row[1]
+    now = dt.datetime.now(tz=dt.UTC)
+    if expires_at is None or expires_at < now:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="reset token expired")
+
+    new_hash = hash_password(body.password)
+    await session.execute(
+        sa_text(
+            "UPDATE users "
+            "SET password_hash = :p, password_reset_token_hash = NULL, "
+            "    password_reset_expires_at = NULL "
+            "WHERE id = :u"
+        ),
+        {"p": new_hash, "u": user_id},
+    )
+    await session.commit()
+
+    # Best-effort revoke of existing sessions; ignore Redis hiccups so the
+    # password itself is still rotated.
+    try:
+        await refresh_store.revoke_all(user_id)
+    except Exception as exc:  # noqa: BLE001
+        log.warning("reset_password_revoke_failed", user_id=str(user_id), error=str(exc))
+
+    log.info("user_password_reset", user_id=str(user_id))
+    return OkResponse(ok=True, detail="password updated")
+
+
+def _require_web_auth_enabled() -> None:
+    """Gate webUI endpoints behind a feature flag (WEB_AUTH_ENABLED)."""
+    if not get_settings().web_auth_enabled:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="web auth disabled")
diff --git a/src/contract_check/core/auth.py b/src/contract_check/core/auth.py
index 78ef311..80aae3b 100644
--- a/src/contract_check/core/auth.py
+++ b/src/contract_check/core/auth.py
@@ -32,6 +32,7 @@ from .logging import get_logger
 log = get_logger(__name__)
 
 JWT_TYPE_ACCESS = "access"
+JWT_TYPE_REFRESH = "refresh"
 
 
 class AuthError(Exception):
@@ -92,6 +93,10 @@ def _jwt_access_ttl() -> dt.timedelta:
     return dt.timedelta(minutes=get_settings().jwt_access_ttl_minutes)
 
 
+def _jwt_refresh_ttl() -> dt.timedelta:
+    return dt.timedelta(days=get_settings().jwt_refresh_ttl_days)
+
+
 def create_access_token(user_id: uuid.UUID, telegram_id: int) -> str:
     """Sign a fresh access JWT for a verified user."""
     settings = get_settings()
@@ -148,6 +153,85 @@ def verify_access_token(token: str) -> AccessTokenClaims:
         raise TokenInvalidError("malformed token claims") from exc
 
 
+@dataclass(slots=True)
+class RefreshTokenClaims:
+    """Payload for a refresh JWT. The `jti` is checked against the refresh store."""
+
+    sub: uuid.UUID  # user_id
+    jti: str  # opaque id used as the Redis-store key
+    type: str
+    exp: int | None = None
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "sub": str(self.sub),
+            "jti": self.jti,
+            "type": self.type,
+            "exp": self.exp,
+        }
+
+    @classmethod
+    def from_dict(cls, payload: dict[str, Any]) -> RefreshTokenClaims:
+        return cls(
+            sub=uuid.UUID(str(payload["sub"])),
+            jti=str(payload["jti"]),
+            type=str(payload.get("type", JWT_TYPE_REFRESH)),
+            exp=payload.get("exp"),
+        )
+
+
+def create_refresh_token(user_id: uuid.UUID, jti: str) -> str:
+    """Sign a refresh JWT. `jti` is the lookup key in the refresh-token store."""
+    settings = get_settings()
+    now = dt.datetime.now(tz=dt.UTC)
+    claims = RefreshTokenClaims(sub=user_id, jti=jti, type=JWT_TYPE_REFRESH)
+    payload = claims.to_dict()
+    payload.update(
+        {
+            "iat": int(now.timestamp()),
+            "exp": int((now + _jwt_refresh_ttl()).timestamp()),
+            "iss": settings.otel_service_name or "contract-check",
+            "aud": "contract-check",
+        }
+    )
+    token: str = jwt.encode(
+        payload,
+        key=_jwt_secret(),
+        algorithm=_jwt_algorithm(),
+    )
+    return token
+
+
+def verify_refresh_token(token: str) -> RefreshTokenClaims:
+    """Verify a refresh JWT signature/expiry. Does NOT check the store — see core.auth_refresh.
+
+    Raises TokenInvalidError / TokenExpiredError on failure.
+    """
+    try:
+        payload = jwt.decode(
+            token,
+            key=_jwt_secret(),
+            algorithms=[_jwt_algorithm()],
+            audience="contract-check",
+            options={
+                "require": ["sub", "jti", "exp", "iat"],
+                "verify_aud": True,
+            },
+        )
+    except jwt.ExpiredSignatureError as exc:
+        raise TokenExpiredError("token expired") from exc
+    except jwt.InvalidTokenError as exc:
+        raise TokenInvalidError("invalid token") from exc
+
+    if payload.get("type") != JWT_TYPE_REFRESH:
+        raise TokenInvalidError("unexpected token type")
+
+    try:
+        return RefreshTokenClaims.from_dict(payload)
+    except (KeyError, ValueError, TypeError) as exc:
+        raise TokenInvalidError("malformed token claims") from exc
+
+
 def _telegram_secret_key(bot_token: str) -> bytes:
     """Telegram uses HMAC_SHA256(BOT_TOKEN, 'WebAppData') as the signing key."""
     return hmac.new(
diff --git a/src/contract_check/core/auth_refresh.py b/src/contract_check/core/auth_refresh.py
new file mode 100644
index 0000000..e42b08e
--- /dev/null
+++ b/src/contract_check/core/auth_refresh.py
@@ -0,0 +1,58 @@
+"""Redis-backed refresh-token store.
+
+Refresh tokens are signed JWTs (core/auth.py) carrying a `jti`. To make logout
+enforceable, the `jti` is also tracked in Redis with a TTL equal to the JWT's
+own expiry. On logout the key is deleted; the access JWT stays valid until its
+short expiry, but no new access tokens can be minted from the revoked refresh.
+
+Keys: `cc:refresh:{user_id}:{jti}` -> "1".
+A Redis flush invalidates every active refresh session (acceptable trade-off
+for stateless access tokens — clients must re-login).
+"""
+
+from __future__ import annotations
+
+import uuid
+from typing import Any
+
+from .auth_refresh_key import refresh_key
+
+
+class RefreshTokenStore:
+    """Wrap a Redis client with typed refresh-token operations."""
+
+    def __init__(self, redis: Any, *, ttl_seconds: int) -> None:
+        self._redis = redis
+        self._ttl = int(ttl_seconds)
+
+    async def issue(self, user_id: uuid.UUID) -> str:
+        """Mint a new jti and persist it with the configured TTL. Returns the jti."""
+        jti = uuid.uuid4().hex
+        await self._redis.set(refresh_key(user_id, jti), "1", ex=self._ttl)
+        return jti
+
+    async def is_valid(self, user_id: uuid.UUID, jti: str) -> bool:
+        """True iff the jti is present (not revoked, not expired)."""
+        if not jti:
+            return False
+        return bool(await self._redis.exists(refresh_key(user_id, jti)))
+
+    async def revoke(self, user_id: uuid.UUID, jti: str) -> bool:
+        """Delete a single jti. Returns True if a key was removed."""
+        if not jti:
+            return False
+        deleted = await self._redis.delete(refresh_key(user_id, jti))
+        return bool(deleted)
+
+    async def revoke_all(self, user_id: uuid.UUID) -> int:
+        """Revoke every active refresh token for a user. Returns count removed."""
+        pattern = refresh_key(user_id, "*")
+        cursor = 0
+        removed = 0
+        while True:
+            cursor, keys = await self._redis.scan(cursor=cursor, match=pattern, count=100)
+            if keys:
+                removed += int(await self._redis.delete(*keys))
+            if int(cursor) == 0:
+                break
+        return removed
diff --git a/src/contract_check/core/auth_refresh_key.py b/src/contract_check/core/auth_refresh_key.py
new file mode 100644
index 0000000..fd00b61
--- /dev/null
+++ b/src/contract_check/core/auth_refresh_key.py
@@ -0,0 +1,15 @@
+"""Key-format helpers for the Redis refresh-token store.
+
+Kept separate from auth_refresh.py so it can be imported by tests/migrations
+without dragging in the redis client type.
+"""
+
+from __future__ import annotations
+
+import uuid
+
+_PREFIX = "cc:refresh"
+
+
+def refresh_key(user_id: uuid.UUID, jti: str) -> str:
+    return f"{_PREFIX}:{user_id}:{jti}"
diff --git a/src/contract_check/core/config.py b/src/contract_check/core/config.py
index 3cfc0a9..8c79321 100644
--- a/src/contract_check/core/config.py
+++ b/src/contract_check/core/config.py
@@ -43,6 +43,7 @@ class Settings(BaseSettings):
     # --- RabbitMQ tuning ---
     mq_prefetch_extract: int = 1
     mq_prefetch_analyze: int = 3
+    mq_prefetch_notify: int = 5
     mq_max_attempts: int = 5
     mq_retry_base_ms: int = 2000
 
@@ -88,6 +89,31 @@ class Settings(BaseSettings):
     jwt_secret: str = Field(..., description="HS256 secret for signing user JWTs")
     jwt_algorithm: str = "HS256"
     jwt_access_ttl_minutes: int = 24 * 60  # 24 hours default; tune per env
+    jwt_refresh_ttl_days: int = 30  # refresh-token lifetime for webUI auth
+
+    # --- webUI auth (email + password) ---
+    web_auth_enabled: bool = Field(
+        default=True,
+        description="Toggle for /api/v1/auth/{register,login,...} email-password routes",
+    )
+    password_reset_ttl_minutes: int = 60  # reset-token validity window
+    web_app_base_url: str = Field(
+        default="http://localhost:5173",
+        description="Base URL of the webUI SPA — used to build password-reset links",
+    )
+    # Minimum password length enforced at register / reset.
+    password_min_length: int = 8
+
+    # --- SMTP (notification transport; empty host disables sending) ---
+    smtp_host: str = ""
+    smtp_port: int = 587
+    smtp_username: str = ""
+    smtp_password: str = ""
+    smtp_from: str = Field(
+        default="no-reply@contract-check.local",
+        description="From: address used by the notify worker",
+    )
+    smtp_use_tls: bool = True  # STARTTLS on port 587; set false for plain SMTP
 
     # --- logging ---
     log_format: str = "json"  # json | console
diff --git a/src/contract_check/core/db/models.py b/src/contract_check/core/db/models.py
index d6e59f1..baacebb 100644
--- a/src/contract_check/core/db/models.py
+++ b/src/contract_check/core/db/models.py
@@ -45,12 +45,26 @@ class User(Base):
         server_default=text("gen_random_uuid()"),
     )
     telegram_id: Mapped[int | None] = mapped_column(BigInteger, unique=True)
+    email: Mapped[str | None] = mapped_column(Text, unique=True)
+    password_hash: Mapped[str | None] = mapped_column(Text)
+    password_reset_token_hash: Mapped[str | None] = mapped_column(Text)
+    password_reset_expires_at: Mapped[dt.datetime | None] = mapped_column(DateTime(timezone=True))
+    is_active: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, default=True, server_default=text("true")
+    )
     created_at: Mapped[dt.datetime] = mapped_column(
         DateTime(timezone=True), nullable=False, server_default=func.now()
     )
     credits_left: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
 
-    __table_args__ = (CheckConstraint("credits_left >= 0", name="users_credits_nonneg"),)
+    __table_args__ = (
+        CheckConstraint("credits_left >= 0", name="users_credits_nonneg"),
+        # A user must have at least one identity anchor.
+        CheckConstraint(
+            "telegram_id IS NOT NULL OR email IS NOT NULL",
+            name="users_identity_present",
+        ),
+    )
 
     documents: Mapped[list[Document]] = relationship(
         back_populates="user", cascade="all, delete-orphan"
diff --git a/src/contract_check/core/metrics.py b/src/contract_check/core/metrics.py
index c0c75ae..8250d6c 100644
--- a/src/contract_check/core/metrics.py
+++ b/src/contract_check/core/metrics.py
@@ -37,7 +37,7 @@ credits_refunded = Counter(
 mq_published = Counter(
     "contract_check_mq_published_total",
     "Messages published to RabbitMQ.",
-    ["queue"],  # extract | analyze
+    ["queue"],  # extract | analyze | notify
 )
 mq_failed = Counter(
     "contract_check_mq_failed_total",
@@ -73,6 +73,10 @@ analyze_duration = Histogram(
     "contract_check_analyze_duration_seconds",
     "Document analysis (analyze.q handler) latency.",
 )
+notify_duration = Histogram(
+    "contract_check_notify_duration_seconds",
+    "Notification send (notify.q handler) latency.",
+)
 
 
 def start_metrics_server(port: int) -> None:
diff --git a/src/contract_check/core/mq/consumer.py b/src/contract_check/core/mq/consumer.py
index 86a9b3c..9d8ceea 100644
--- a/src/contract_check/core/mq/consumer.py
+++ b/src/contract_check/core/mq/consumer.py
@@ -22,12 +22,12 @@ import asyncio
 from typing import TYPE_CHECKING, Any
 
 from aio_pika import DeliveryMode, Message, connect_robust
-from pydantic import ValidationError
+from pydantic import BaseModel, ValidationError
 
 from ..db.enums import FailureClass
 from ..errors import TerminalError
 from ..logging import get_logger, set_correlation_id
-from .messages import PipelineMessage
+from .messages import NotificationMessage, PipelineMessage
 from .topology import (
     DLQ_FOR,
     EXCHANGE_RETRY,
@@ -49,18 +49,27 @@ if TYPE_CHECKING:
 log = get_logger(__name__)
 
 
+# Any pydantic BaseModel with an `attempt` int field and a `next_attempt()`
+# method can ride the retry mechanics. PipelineMessage and NotificationMessage
+# both qualify.
+type RetryableMessage = PipelineMessage | NotificationMessage
+
+
 def _is_terminal(exc: BaseException) -> bool:
     """Return True for errors that will not be fixed by RabbitMQ retries."""
     return isinstance(exc, TerminalError)
 
 
-class Consumer[MsgT: PipelineMessage]:
+class Consumer[MsgT: RetryableMessage]:
     """Base RabbitMQ consumer bound to one main queue."""
 
     #: subclass declares which queue + routing key + message model it owns
     queue: str = ""
     routing_key: str = ""
-    message_model: type[PipelineMessage] = PipelineMessage
+    message_model: type[BaseModel] = PipelineMessage
+    #: retry exchange used when nacking (defaults to the contracts retry exchange;
+    #: notify pipeline overrides with EXCHANGE_NOTIFY_RETRY).
+    retry_exchange: str = EXCHANGE_RETRY
 
     def __init__(
         self,
@@ -229,7 +238,7 @@ class Consumer[MsgT: PipelineMessage]:
             H_ATTEMPT: new_attempt,
             H_ORIGIN: self._origin,
         }
-        retry_exchange = await self._channel.get_exchange(EXCHANGE_RETRY, ensure=False)
+        retry_exchange = await self._channel.get_exchange(self.retry_exchange, ensure=False)
         await retry_exchange.publish(
             Message(
                 body,
diff --git a/src/contract_check/core/mq/messages.py b/src/contract_check/core/mq/messages.py
index dd00c86..b58036b 100644
--- a/src/contract_check/core/mq/messages.py
+++ b/src/contract_check/core/mq/messages.py
@@ -9,7 +9,7 @@ traces trace api → rabbit → workers under one id.
 from __future__ import annotations
 
 import uuid
-from typing import Self
+from typing import Literal, Self
 
 from pydantic import BaseModel, Field
 
@@ -47,3 +47,32 @@ class DocumentExtracted(PipelineMessage):
     extracted_s3_key: str
     char_count: int
     ocr_used: bool
+
+
+# ── notifications ──────────────────────────────────────────────────────────────
+# The notify pipeline is decoupled from the contracts pipeline: it shares the
+# retry/DLQ mechanics but lives on its own exchange/queue (see topology.py).
+NotificationKind = Literal[
+    "password_reset",
+    "welcome",
+    "email_verification",
+]
+
+
+class NotificationMessage(BaseModel):
+    """api → notify.x[notify] → worker-notify.
+
+    Carries everything the SMTP transport needs so the worker is stateless
+    beyond the broker. `context` is a free-form payload interpreted per `kind`.
+    """
+
+    correlation_id: uuid.UUID
+    kind: NotificationKind
+    to: str = Field(..., description="Recipient email address")
+    subject: str
+    body_text: str
+    body_html: str | None = None
+    attempt: int = Field(default=0, ge=0)
+
+    def next_attempt(self) -> Self:
+        return self.model_copy(update={"attempt": self.attempt + 1})
diff --git a/src/contract_check/core/mq/topology.py b/src/contract_check/core/mq/topology.py
index dc67f42..da756b2 100644
--- a/src/contract_check/core/mq/topology.py
+++ b/src/contract_check/core/mq/topology.py
@@ -36,6 +36,8 @@ if TYPE_CHECKING:
 # ── exchanges ────────────────────────────────────────────────────────────────
 EXCHANGE_MAIN = "contracts.x"
 EXCHANGE_RETRY = "contracts.retry.x"
+EXCHANGE_NOTIFY = "notify.x"
+EXCHANGE_NOTIFY_RETRY = "notify.retry.x"
 
 # ── queues ───────────────────────────────────────────────────────────────────
 QUEUE_EXTRACT = "extract.q"
@@ -44,12 +46,17 @@ QUEUE_EXTRACT_RETRY = "extract.retry.q"
 QUEUE_ANALYZE_RETRY = "analyze.retry.q"
 QUEUE_EXTRACT_DLQ = "extract.dlq"
 QUEUE_ANALYZE_DLQ = "analyze.dlq"
+QUEUE_NOTIFY = "notify.q"
+QUEUE_NOTIFY_RETRY = "notify.retry.q"
+QUEUE_NOTIFY_DLQ = "notify.dlq"
 
 # ── routing keys ─────────────────────────────────────────────────────────────
 RK_EXTRACT = "extract"
 RK_ANALYZE = "analyze"
 RK_RETRY_EXTRACT = "retry.extract"
 RK_RETRY_ANALYZE = "retry.analyze"
+RK_NOTIFY = "notify"
+RK_RETRY_NOTIFY = "retry.notify"
 
 # ── message headers ──────────────────────────────────────────────────────────
 H_CORRELATION_ID = "x-correlation-id"
@@ -70,6 +77,7 @@ _RETRY_QUEUE_FOR: dict[str, tuple[str, str, str]] = {
 DLQ_FOR: dict[str, str] = {
     QUEUE_EXTRACT: QUEUE_EXTRACT_DLQ,
     QUEUE_ANALYZE: QUEUE_ANALYZE_DLQ,
+    QUEUE_NOTIFY: QUEUE_NOTIFY_DLQ,
 }
 
 
@@ -115,9 +123,34 @@ async def declare_all(channel: AbstractChannel) -> None:
         await rq.bind(retry_exchange, routing_key=f"retry.{main_rk}")
 
     # Quorum DLQs (no bindings; published to directly via default exchange).
-    for dlq in (QUEUE_EXTRACT_DLQ, QUEUE_ANALYZE_DLQ):
+    for dlq in (QUEUE_EXTRACT_DLQ, QUEUE_ANALYZE_DLQ, QUEUE_NOTIFY_DLQ):
         await channel.declare_queue(dlq, durable=True, arguments=_QUORUM_ARGS)
 
+    # ── notify pipeline (parallel to the contracts pipeline) ────────────────
+    notify_exchange = await channel.declare_exchange(EXCHANGE_NOTIFY, durable=True)
+    notify_retry_exchange = await channel.declare_exchange(EXCHANGE_NOTIFY_RETRY, durable=True)
+
+    notify_q = await channel.declare_queue(
+        QUEUE_NOTIFY,
+        durable=True,
+        arguments={
+            **_QUORUM_ARGS,
+            "x-dead-letter-exchange": EXCHANGE_NOTIFY_RETRY,
+            "x-dead-letter-routing-key": RK_RETRY_NOTIFY,
+        },
+    )
+    await notify_q.bind(notify_exchange, routing_key=RK_NOTIFY)
+
+    notify_retry_q = await channel.declare_queue(
+        QUEUE_NOTIFY_RETRY,
+        durable=True,
+        arguments={
+            "x-dead-letter-exchange": EXCHANGE_NOTIFY,
+            "x-dead-letter-routing-key": RK_NOTIFY,
+        },
+    )
+    await notify_retry_q.bind(notify_retry_exchange, routing_key=RK_RETRY_NOTIFY)
+
 
 # ── lazy-mode policy for existing (non-lazy) retry queues ───────────────────
 # RabbitMQ classic queues only expire per-message TTL when the message reaches
diff --git a/src/contract_check/core/notifications/__init__.py b/src/contract_check/core/notifications/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/src/contract_check/core/notifications/publisher.py b/src/contract_check/core/notifications/publisher.py
new file mode 100644
index 0000000..3d1203c
--- /dev/null
+++ b/src/contract_check/core/notifications/publisher.py
@@ -0,0 +1,86 @@
+"""RabbitMQ publisher for the notify pipeline.
+
+Bound to the `notify.x` exchange (separate from contracts.x so the two domains
+have independent DLQ/retry topology). Same publisher-confirms semantics as the
+contracts Publisher: a publish that is not confirmed raises, so the api never
+returns 202/200 with a notification silently dropped.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from aio_pika import DeliveryMode, Message, connect_robust
+
+from ..logging import get_logger
+from ..mq.messages import NotificationMessage
+from ..mq.topology import (
+    EXCHANGE_NOTIFY,
+    H_ATTEMPT,
+    H_CORRELATION_ID,
+    H_ORIGIN,
+    declare_all,
+)
+
+if TYPE_CHECKING:
+    from aio_pika.abc import AbstractChannel, AbstractExchange, AbstractRobustConnection
+
+log = get_logger(__name__)
+
+
+class NotificationPublisher:
+    """Robust RabbitMQ publisher bound to the notify exchange."""
+
+    def __init__(self, url: str, *, origin: str = "api") -> None:
+        self._url = url
+        self._origin = origin
+        self._conn: AbstractRobustConnection | None = None
+        self._channel: AbstractChannel | None = None
+        self._exchange: AbstractExchange | None = None
+
+    async def connect(self) -> None:
+        self._conn = await connect_robust(self._url)
+        channel = await self._conn.channel(publisher_confirms=True)
+        self._channel = channel
+        # declare_all is idempotent and also declares the notify topology.
+        await declare_all(channel)
+        self._exchange = await channel.declare_exchange(EXCHANGE_NOTIFY, durable=True)
+        log.info("notification_publisher_connected", url=self._url, origin=self._origin)
+
+    async def publish(self, message: NotificationMessage, routing_key: str = "notify") -> None:
+        """Publish a NotificationMessage. Raises on no-confirm."""
+        if self._exchange is None:
+            raise RuntimeError("NotificationPublisher not connected; call connect() first")
+        body = message.model_dump_json().encode("utf-8")
+        amqp = Message(
+            body,
+            content_type="application/json",
+            delivery_mode=DeliveryMode.PERSISTENT,
+            correlation_id=str(message.correlation_id),
+            message_id=str(message.correlation_id),
+            headers={
+                H_CORRELATION_ID: str(message.correlation_id),
+                H_ATTEMPT: message.attempt,
+                H_ORIGIN: self._origin,
+            },
+        )
+        await self._exchange.publish(amqp, routing_key=routing_key)
+        log.debug(
+            "notification_published",
+            routing_key=routing_key,
+            correlation_id=str(message.correlation_id),
+            kind=message.kind,
+            to=message.to,
+        )
+
+    async def close(self) -> None:
+        if self._conn is not None:
+            await self._conn.close()
+        self._conn = self._channel = self._exchange = None
+
+    async def __aenter__(self) -> NotificationPublisher:
+        await self.connect()
+        return self
+
+    async def __aexit__(self, *exc: object) -> None:
+        await self.close()
diff --git a/src/contract_check/core/notifications/transport.py b/src/contract_check/core/notifications/transport.py
new file mode 100644
index 0000000..b05616d
--- /dev/null
+++ b/src/contract_check/core/notifications/transport.py
@@ -0,0 +1,88 @@
+"""Async SMTP transport for the notify worker.
+
+Kept as a thin port so tests can substitute a no-op sink. `aiosmtplib` is the
+only async sender; in dev (SMTP_HOST empty) we just log the message body.
+"""
+
+from __future__ import annotations
+
+import email.message
+from typing import Protocol
+
+import aiosmtplib
+
+from ..config import Settings, get_settings
+from ..logging import get_logger
+from ..mq.messages import NotificationMessage
+
+log = get_logger(__name__)
+
+
+class NotificationTransport(Protocol):
+    """Send one notification. Implementations: SMTP / dev-logger."""
+
+    async def send(self, message: NotificationMessage) -> None: ...
+
+
+class DevLogTransport:
+    """Dev fallback: log instead of sending. Used when SMTP_HOST is empty."""
+
+    async def send(self, message: NotificationMessage) -> None:
+        log.info(
+            "notification_dev_sink",
+            kind=message.kind,
+            to=message.to,
+            subject=message.subject,
+            body=message.body_text,
+        )
+
+
+class SmtpTransport:
+    """Production SMTP transport via aiosmtplib. Connections are per-send to keep the
+    worker stateless; for high throughput swap to a pooled connection.
+    """
+
+    def __init__(self, settings: Settings | None = None) -> None:
+        self._settings = settings or get_settings()
+
+    async def send(self, message: NotificationMessage) -> None:
+        settings = self._settings
+        if not settings.smtp_host:
+            # Misconfiguration at runtime — fall back to logging rather than crashing.
+            log.warning(
+                "smtp_host_empty_fallback_to_dev",
+                kind=message.kind,
+                to=message.to,
+            )
+            await DevLogTransport().send(message)
+            return
+
+        msg = email.message.EmailMessage()
+        msg["From"] = settings.smtp_from
+        msg["To"] = message.to
+        msg["Subject"] = message.subject
+        msg.set_content(message.body_text)
+        if message.body_html:
+            msg.add_alternative(message.body_html, subtype="html")
+
+        try:
+            await aiosmtplib.send(
+                msg,
+                hostname=settings.smtp_host,
+                port=settings.smtp_port,
+                username=settings.smtp_username or None,
+                password=settings.smtp_password or None,
+                start_tls=settings.smtp_use_tls,
+            )
+            log.info(
+                "notification_sent",
+                kind=message.kind,
+                to=message.to,
+                subject=message.subject,
+            )
+        except Exception as exc:  # noqa: BLE001 — surfaced as retryable infra failure
+            raise SmtpTransportError(f"smtp send failed: {exc}") from exc
+
+
+class SmtpTransportError(Exception):
+    """Raised when SMTP delivery fails so the consumer can retry/DLQ."""
diff --git a/src/contract_check/core/security/__init__.py b/src/contract_check/core/security/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/src/contract_check/core/security/passwords.py b/src/contract_check/core/security/passwords.py
new file mode 100644
index 0000000..801562f
--- /dev/null
+++ b/src/contract_check/core/security/passwords.py
@@ -0,0 +1,55 @@
+"""Password hashing (argon2id) via argon2-cffi.
+
+Pure crypto helpers — no FastAPI, no DB. Used by the api at register/login and
+by password-reset. argon2-cffi is the OWASP-recommended PHC-format hasher; the
+encoded hash embeds salt + params so the verifier auto-detects them on verify.
+"""
+
+from __future__ import annotations
+
+from argon2 import PasswordHasher
+from argon2.exceptions import InvalidHash, VerificationError, VerifyMismatchError
+
+# Reasonable defaults (OWASP cheat-sheet). Tunable via env on the hasher if needed.
+_hasher = PasswordHasher(
+    time_cost=3,
+    memory_cost=64 * 1024,  # 64 MiB
+    parallelism=2,
+)
+
+
+class PasswordError(Exception):
+    """Raised when a password fails hashing or verification."""
+
+
+def hash_password(plain: str) -> str:
+    """Return an argon2id PHC-string hash of `plain`. Raises PasswordError on misuse."""
+    if not plain:
+        raise PasswordError("password must not be empty")
+    try:
+        return _hasher.hash(plain)
+    except (ValueError, TypeError) as exc:
+        raise PasswordError(f"failed to hash password: {exc}") from exc
+
+
+def verify_password(plain: str, hashed: str) -> bool:
+    """Constant-time check that `plain` matches a stored PHC hash.
+
+    Returns False on mismatch or malformed hash (never raises for those cases).
+    """
+    if not plain or not hashed:
+        return False
+    try:
+        return _hasher.verify(hashed, plain)
+    except VerifyMismatchError:
+        return False
+    except (VerificationError, InvalidHash):
+        return False
+
+
+def needs_rehash(hashed: str) -> bool:
+    """True if the stored hash uses outdated params and should be re-hashed on next login."""
+    try:
+        return _hasher.check_needs_rehash(hashed)
+    except (InvalidHash, TypeError):
+        return False
diff --git a/src/contract_check/worker_notify/__init__.py b/src/contract_check/worker_notify/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/src/contract_check/worker_notify/__main__.py b/src/contract_check/worker_notify/__main__.py
new file mode 100644
index 0000000..0859c8b
--- /dev/null
+++ b/src/contract_check/worker_notify/__main__.py
@@ -0,0 +1,62 @@
+"""worker-notify entrypoint: connects to RabbitMQ and runs the notify consumer."""
+
+from __future__ import annotations
+
+import asyncio
+import signal
+
+from ..core.config import get_settings
+from ..core.logging import bind_context, configure_logging, get_logger
+from ..core.metrics import start_metrics_server
+from ..core.sentry import init_sentry
+from ..core.telemetry import setup_telemetry, shutdown_telemetry
+from .consumer import NotifyConsumer
+
+log = get_logger(__name__)
+
+
+async def main() -> None:
+    settings = get_settings()
+    configure_logging(
+        settings.log_level,
+        json_output=settings.json_logs,
+        service="worker-notify",
+        env=settings.env,
+    )
+    bind_context(service="worker-notify", env=settings.env)
+    init_sentry("worker-notify")
+    setup_telemetry("worker-notify")
+
+    start_metrics_server(9103)
+
+    consumer = NotifyConsumer(
+        url=settings.rabbitmq_url,
+        origin="worker-notify",
+        prefetch=settings.mq_prefetch_notify,
+        max_attempts=settings.mq_max_attempts,
+        retry_base_ms=settings.mq_retry_base_ms,
+    )
+    await consumer.connect()
+
+    loop = asyncio.get_running_loop()
+    stop_event = asyncio.Event()
+
+    for sig in (signal.SIGINT, signal.SIGTERM):
+        loop.add_signal_handler(sig, stop_event.set)
+
+    consumer_task = asyncio.create_task(consumer.run())
+    stop_task = asyncio.create_task(stop_event.wait())
+
+    log.info("worker_notify_started", prefetch=settings.mq_prefetch_notify)
+    try:
+        await asyncio.wait(
+            {consumer_task, stop_task},
+            return_when=asyncio.FIRST_COMPLETED,
+        )
+    finally:
+        await consumer.stop()
+        shutdown_telemetry()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
diff --git a/src/contract_check/worker_notify/consumer.py b/src/contract_check/worker_notify/consumer.py
new file mode 100644
index 0000000..73858c3
--- /dev/null
+++ b/src/contract_check/worker_notify/consumer.py
@@ -0,0 +1,61 @@
+"""worker-notify consumer: wires the notify handler into the RabbitMQ base.
+
+Consumes `notify.q` and routes failures to `notify.retry.q` / `notify.dlq`
+via the topology declared in core/mq/topology.py.
+"""
+
+from __future__ import annotations
+
+from ..core.db.enums import FailureClass
+from ..core.logging import get_logger
+from ..core.metrics import notify_duration
+from ..core.mq.consumer import Consumer
+from ..core.mq.messages import NotificationMessage
+from ..core.mq.topology import EXCHANGE_NOTIFY_RETRY
+from .handler import NotifyHandler
+
+log = get_logger(__name__)
+
+
+class NotifyConsumer(Consumer[NotificationMessage]):
+    """Consumes `notify.q` and dispatches via the SMTP transport."""
+
+    queue: str = "notify.q"
+    routing_key: str = "notify"
+    message_model = NotificationMessage
+    retry_exchange: str = EXCHANGE_NOTIFY_RETRY
+
+    def __init__(
+        self,
+        url: str,
+        *,
+        origin: str,
+        prefetch: int,
+        max_attempts: int,
+        retry_base_ms: int,
+    ) -> None:
+        super().__init__(
+            url,
+            origin=origin,
+            prefetch=prefetch,
+            max_attempts=max_attempts,
+            retry_base_ms=retry_base_ms,
+        )
+        self._handler = NotifyHandler()
+
+    def classify(self, exc: BaseException) -> FailureClass:
+        return self._handler.classify(exc)
+
+    @notify_duration.time()
+    async def handle(self, payload: NotificationMessage) -> None:
+        await self._handler.handle(payload)
+
+    async def on_failure(
+        self, payload: NotificationMessage, failure_class: FailureClass, attempt: int, error: str
+    ) -> None:
+        await self._handler.on_failure(payload, failure_class, attempt, error)
+
+    async def on_dlq(
+        self, payload: NotificationMessage, failure_class: FailureClass, error: str
+    ) -> None:
+        await self._handler.on_terminal_failure(payload, failure_class, error)
diff --git a/src/contract_check/worker_notify/handler.py b/src/contract_check/worker_notify/handler.py
new file mode 100644
index 0000000..4ea90d0
--- /dev/null
+++ b/src/contract_check/worker_notify/handler.py
@@ -0,0 +1,78 @@
+"""worker-notify handler: deliver one NotificationMessage via the SMTP transport.
+
+Kept separate from the consumer so it can be unit-tested in-process without a
+real RabbitMQ. Owns:
+  - transport selection (SMTP in prod, dev logger when SMTP_HOST is empty)
+  - failure classification (smtp failures → infra, retried by the consumer)
+"""
+
+from __future__ import annotations
+
+from ..core.config import get_settings
+from ..core.db.enums import FailureClass
+from ..core.logging import get_logger
+from ..core.metrics import mq_failed
+from ..core.mq.messages import NotificationMessage
+from ..core.notifications.transport import (
+    DevLogTransport,
+    NotificationTransport,
+    SmtpTransport,
+)
+
+log = get_logger(__name__)
+
+
+class NotifyHandler:
+    """Business logic for worker-notify."""
+
+    def __init__(self, transport: NotificationTransport | None = None) -> None:
+        self._settings = get_settings()
+        self._transport: NotificationTransport = transport or self._default_transport()
+
+    def _default_transport(self) -> NotificationTransport:
+        # Dev sink when SMTP is not configured; production uses real SMTP.
+        if not self._settings.smtp_host:
+            return DevLogTransport()
+        return SmtpTransport(self._settings)
+
+    async def handle(self, payload: NotificationMessage) -> None:
+        await self._transport.send(payload)
+        log.info(
+            "notify_delivered",
+            kind=payload.kind,
+            to=payload.to,
+            attempt=payload.attempt,
+        )
+
+    def classify(self, exc: BaseException) -> FailureClass:
+        # Any SMTP-side failure is treated as infra (transient network/cred).
+        from ..core.notifications.transport import SmtpTransportError
+
+        if isinstance(exc, SmtpTransportError):
+            return "infra"
+        return "unknown"
+
+    async def on_failure(
+        self, payload: NotificationMessage, failure_class: FailureClass, attempt: int, error: str
+    ) -> None:
+        mq_failed.labels(queue="notify", failure_class=failure_class).inc()
+        log.warning(
+            "notify_failed",
+            kind=payload.kind,
+            to=payload.to,
+            attempt=attempt,
+            failure_class=failure_class,
+            error=error,
+        )
+
+    async def on_terminal_failure(
+        self, payload: NotificationMessage, failure_class: FailureClass, error: str
+    ) -> None:
+        mq_failed.labels(queue="notify", failure_class=failure_class).inc()
+        log.error(
+            "notify_dlq",
+            kind=payload.kind,
+            to=payload.to,
+            failure_class=failure_class,
+            error=error,
+        )
diff --git a/srv/worker-notify/Dockerfile b/srv/worker-notify/Dockerfile
new file mode 100644
index 0000000..d380bcb
--- /dev/null
+++ b/srv/worker-notify/Dockerfile
@@ -0,0 +1,34 @@
+# syntax=docker/dockerfile:1
+
+# ─── Stage 1: build deps + project into a venv via uv ─────────────────────────
+FROM ghcr.io/astral-sh/uv:python3.13-bookworm-slim AS builder
+
+ENV UV_COMPILE_BYTECODE=1 \
+    UV_LINK_MODE=copy \
+    UV_PYTHON_DOWNLOADS=never \
+    UV_PROJECT_ENVIRONMENT=/app/.venv
+
+WORKDIR /app
+
+COPY pyproject.toml uv.lock ./
+COPY README.md ./
+COPY src ./src
+
+# Install only the notify group (core + db/mq/obs + aiosmtplib).
+RUN --mount=type=cache,target=/root/.cache/uv \
+    uv sync --frozen --no-default-groups --group notify --no-install-project && \
+    uv pip install --no-deps .
+
+# ─── Stage 2: lean runtime ─────────────────────────────────────────────────
+FROM python:3.13-slim
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    PATH=/app/.venv/bin:$PATH
+
+WORKDIR /app
+
+COPY --from=builder /app/.venv /app/.venv
+
+EXPOSE 9103
+CMD ["python", "-m", "contract_check.worker_notify"]
diff --git a/tests/unit/test_web_auth.py b/tests/unit/test_web_auth.py
new file mode 100644
index 0000000..77cc382
--- /dev/null
+++ b/tests/unit/test_web_auth.py
@@ -0,0 +1,223 @@
+"""Unit tests for webUI auth: argon2 hashing, refresh-token store, JWT pair."""
+
+from __future__ import annotations
+
+import datetime as dt
+import uuid
+from collections.abc import Callable
+
+import pytest
+
+from contract_check.core.auth import (
+    RefreshTokenClaims,
+    TokenExpiredError,
+    TokenInvalidError,
+    create_refresh_token,
+    verify_refresh_token,
+)
+from contract_check.core.auth_refresh import RefreshTokenStore
+from contract_check.core.auth_refresh_key import refresh_key
+from contract_check.core.config import get_settings
+from contract_check.core.security.passwords import (
+    PasswordError,
+    hash_password,
+    needs_rehash,
+    verify_password,
+)
+
+
+@pytest.fixture(autouse=True)
+def _clear_settings_cache(monkeypatch: pytest.MonkeyPatch) -> None:
+    get_settings.cache_clear()
+    monkeypatch.setenv("JWT_SECRET", "unit-test-secret-do-not-use-in-prod")
+    monkeypatch.setenv("JWT_ALGORITHM", "HS256")
+    monkeypatch.setenv("JWT_ACCESS_TTL_MINUTES", "1440")
+    monkeypatch.setenv("JWT_REFRESH_TTL_DAYS", "30")
+    get_settings.cache_clear()
+    yield
+    get_settings.cache_clear()
+
+
+# ── argon2 ────────────────────────────────────────────────────────────────────
+
+
+def test_hash_and_verify_password_roundtrip() -> None:
+    h = hash_password("correct horse battery staple")
+    assert h != "correct horse battery staple"
+    assert h.startswith("$argon2id$")
+    assert verify_password("correct horse battery staple", h)
+
+
+def test_verify_password_rejects_wrong_password() -> None:
+    h = hash_password("hunter2")
+    assert not verify_password("hunter3", h)
+    assert not verify_password("", h)
+    assert not verify_password("hunter2", "")
+
+
+def test_hash_password_rejects_empty() -> None:
+    with pytest.raises(PasswordError):
+        hash_password("")
+
+
+def test_verify_password_handles_malformed_hash() -> None:
+    assert not verify_password("anything", "not-a-real-hash")
+    assert not verify_password("anything", "$argon2id$truncated")
+
+
+def test_needs_rehash_returns_false_for_fresh_hash() -> None:
+    h = hash_password("supersecret")
+    assert needs_rehash(h) is False
+
+
+# ── refresh JWT ───────────────────────────────────────────────────────────────
+
+
+def test_create_and_verify_refresh_token() -> None:
+    user_id = uuid.uuid4()
+    jti = uuid.uuid4().hex
+    token = create_refresh_token(user_id, jti)
+    claims = verify_refresh_token(token)
+    assert claims.sub == user_id
+    assert claims.jti == jti
+    assert claims.type == "refresh"
+
+
+def test_verify_refresh_rejects_access_token() -> None:
+    from contract_check.core.auth import create_access_token
+
+    access = create_access_token(uuid.uuid4(), 42)
+    with pytest.raises(TokenInvalidError):
+        verify_refresh_token(access)
+
+
+def test_verify_refresh_rejects_tampered_token() -> None:
+    token = create_refresh_token(uuid.uuid4(), uuid.uuid4().hex)
+    tampered = token[:-10] + ("A" if token[-10] != "A" else "B") + token[-9:]
+    with pytest.raises(TokenInvalidError):
+        verify_refresh_token(tampered)
+
+
+def test_verify_refresh_rejects_expired(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("JWT_REFRESH_TTL_DAYS", "-1")
+    get_settings.cache_clear()
+    token = create_refresh_token(uuid.uuid4(), uuid.uuid4().hex)
+    with pytest.raises(TokenExpiredError):
+        verify_refresh_token(token)
+
+
+def test_refresh_claims_roundtrip() -> None:
+    c = RefreshTokenClaims(sub=uuid.uuid4(), jti="abc", type="refresh", exp=123)
+    d = c.to_dict()
+    c2 = RefreshTokenClaims.from_dict(d)
+    assert c2.sub == c.sub
+    assert c2.jti == c.jti
+    assert c2.exp == c.exp
+
+
+# ── refresh-token Redis store ─────────────────────────────────────────────────
+
+
+class _FakeRedis:
+    """In-memory async stand-in for redis.asyncio.Redis (SET/GET/DELETE/EXISTS/SCAN)."""
+
+    def __init__(self) -> None:
+        self._data: dict[str, str] = {}
+        self._ttls: dict[str, float] = {}
+
+    async def set(self, key: str, value: str, ex: int | None = None) -> None:
+        self._data[key] = value
+        if ex is not None:
+            self._ttls[key] = dt.datetime.now(tz=dt.UTC).timestamp() + ex
+
+    async def exists(self, key: str) -> int:
+        if key in self._data and self._unexpired(key):
+            return 1
+        return 0
+
+    async def delete(self, *keys: str) -> int:
+        removed = 0
+        for k in keys:
+            if k in self._data:
+                del self._data[k]
+                self._ttls.pop(k, None)
+                removed += 1
+        return removed
+
+    async def scan(
+        self, cursor: int = 0, match: str | None = None, count: int = 100
+    ) -> tuple[int, list[str]]:
+        import fnmatch
+
+        all_keys = [k for k in self._data if self._unexpired(k)]
+        if match:
+            all_keys = [k for k in all_keys if fnmatch.fnmatch(k, match)]
+        return 0, all_keys
+
+    def _unexpired(self, key: str) -> bool:
+        if key not in self._ttls:
+            return True
+        return dt.datetime.now(tz=dt.UTC).timestamp() < self._ttls[key]
+
+
+@pytest.fixture
+def store_factory() -> Callable[[], tuple[RefreshTokenStore, _FakeRedis]]:
+    def _make() -> tuple[RefreshTokenStore, _FakeRedis]:
+        redis = _FakeRedis()
+        return RefreshTokenStore(redis, ttl_seconds=3600), redis
+
+    return _make
+
+
+async def test_refresh_store_issue_and_validate(
+    store_factory: Callable[[], tuple[RefreshTokenStore, _FakeRedis]],
+) -> None:
+    store, redis = store_factory()
+    user_id = uuid.uuid4()
+
+    jti = await store.issue(user_id)
+
+    assert isinstance(jti, str)
+    assert len(jti) == 32  # uuid4().hex
+    assert await store.is_valid(user_id, jti) is True
+    assert refresh_key(user_id, jti) in redis._data
+
+
+async def test_refresh_store_revoke(
+    store_factory: Callable[[], tuple[RefreshTokenStore, _FakeRedis]],
+) -> None:
+    store, _ = store_factory()
+    user_id = uuid.uuid4()
+
+    jti = await store.issue(user_id)
+    assert await store.revoke(user_id, jti) is True
+    assert await store.is_valid(user_id, jti) is False
+    # Idempotent revoke.
+    assert await store.revoke(user_id, jti) is False
+
+
+async def test_refresh_store_revoke_all(
+    store_factory: Callable[[], tuple[RefreshTokenStore, _FakeRedis]],
+) -> None:
+    store, _ = store_factory()
+    user_id = uuid.uuid4()
+    other = uuid.uuid4()
+
+    jti1 = await store.issue(user_id)
+    jti2 = await store.issue(user_id)
+    jti_other = await store.issue(other)
+
+    removed = await store.revoke_all(user_id)
+
+    assert removed == 2
+    assert await store.is_valid(user_id, jti1) is False
+    assert await store.is_valid(user_id, jti2) is False
+    # Other user unaffected.
+    assert await store.is_valid(other, jti_other) is True
+
+
+async def test_refresh_store_is_valid_rejects_empty_jti(
+    store_factory: Callable[[], tuple[RefreshTokenStore, _FakeRedis]],
+) -> None:
+    store, _ = store_factory()
+    assert await store.is_valid(uuid.uuid4(), "") is False
diff --git a/uv.lock b/uv.lock
index bf021bb..d3157d6 100644
--- a/uv.lock
+++ b/uv.lock
@@ -161,6 +161,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/fb/76/641ae371508676492379f16e2fa48f4e2c11741bd63c48be4b12a6b09cba/aiosignal-1.4.0-py3-none-any.whl", hash = "sha256:053243f8b92b990551949e63930a839ff0cf0b0ebbe0597b0f3fb19e1a0fe82e", size = 7490, upload-time = "2025-07-03T22:54:42.156Z" },
 ]
 
+[[package]]
+name = "aiosmtplib"
+version = "5.1.2"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/b9/25/d36d056e62a1dc3dd51ce76e7647c62966702193dc91eafa7fd4e1006a91/aiosmtplib-5.1.2.tar.gz", hash = "sha256:04a0ea3c678f5b719f998f290dce010ca512e1385836d3944206299df03b060f", size = 71031, upload-time = "2026-06-20T15:00:48.269Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/d5/ec/c5a415cd1309eaac28ad3c599458194d25ba07189bb07a1bac2c6713c17e/aiosmtplib-5.1.2-py3-none-any.whl", hash = "sha256:070d467cc329dafd0af59108ba5d217d973cba10309910fed359a2a7bfb52d7a", size = 28394, upload-time = "2026-06-20T15:00:47.299Z" },
+]
+
 [[package]]
 name = "aiosqlite"
 version = "0.22.1"
@@ -571,7 +580,9 @@ analyze = [
 api = [
     { name = "aio-pika" },
     { name = "alembic" },
+    { name = "argon2-cffi" },
     { name = "asyncpg" },
+    { name = "email-validator" },
     { name = "fastapi" },
     { name = "minio" },
     { name = "opentelemetry-exporter-otlp" },
@@ -598,11 +609,14 @@ db = [
 dev = [
     { name = "aio-pika" },
     { name = "aiogram" },
+    { name = "aiosmtplib" },
     { name = "aiosqlite" },
     { name = "alembic" },
     { name = "anyio" },
+    { name = "argon2-cffi" },
     { name = "asgi-lifespan" },
     { name = "asyncpg" },
+    { name = "email-validator" },
     { name = "fastapi" },
     { name = "minio" },
     { name = "mypy" },
@@ -646,6 +660,17 @@ extract = [
 mq = [
     { name = "aio-pika" },
 ]
+notify = [
+    { name = "aio-pika" },
+    { name = "aiosmtplib" },
+    { name = "alembic" },
+    { name = "asyncpg" },
+    { name = "opentelemetry-exporter-otlp" },
+    { name = "opentelemetry-sdk" },
+    { name = "prometheus-client" },
+    { name = "sentry-sdk" },
+    { name = "sqlalchemy" },
+]
 obs = [
     { name = "opentelemetry-exporter-otlp" },
     { name = "opentelemetry-sdk" },
@@ -685,7 +710,9 @@ analyze = [
 api = [
     { name = "aio-pika", specifier = ">=9.4" },
     { name = "alembic", specifier = ">=1.13" },
+    { name = "argon2-cffi", specifier = ">=23.1" },
     { name = "asyncpg", specifier = ">=0.29" },
+    { name = "email-validator", specifier = ">=2.1" },
     { name = "fastapi", specifier = ">=0.110" },
     { name = "minio", specifier = ">=7.2" },
     { name = "opentelemetry-exporter-otlp", specifier = ">=1.24" },
@@ -710,11 +737,14 @@ db = [
 dev = [
     { name = "aio-pika", specifier = ">=9.4" },
     { name = "aiogram", specifier = ">=3.4" },
+    { name = "aiosmtplib", specifier = ">=3.0" },
     { name = "aiosqlite", specifier = ">=0.20" },
     { name = "alembic", specifier = ">=1.13" },
     { name = "anyio", specifier = ">=4" },
+    { name = "argon2-cffi", specifier = ">=23.1" },
     { name = "asgi-lifespan", specifier = ">=2.1.0" },
     { name = "asyncpg", specifier = ">=0.29" },
+    { name = "email-validator", specifier = ">=2.1" },
     { name = "fastapi", specifier = ">=0.110" },
     { name = "minio", specifier = ">=7.2" },
     { name = "mypy", specifier = ">=1.10" },
@@ -756,6 +786,17 @@ extract = [
     { name = "sqlalchemy", specifier = ">=2.0" },
 ]
 mq = [{ name = "aio-pika", specifier = ">=9.4" }]
+notify = [
+    { name = "aio-pika", specifier = ">=9.4" },
+    { name = "aiosmtplib", specifier = ">=3.0" },
+    { name = "alembic", specifier = ">=1.13" },
+    { name = "asyncpg", specifier = ">=0.29" },
+    { name = "opentelemetry-exporter-otlp", specifier = ">=1.24" },
+    { name = "opentelemetry-sdk", specifier = ">=1.24" },
+    { name = "prometheus-client", specifier = ">=0.20" },
+    { name = "sentry-sdk", specifier = ">=2" },
+    { name = "sqlalchemy", specifier = ">=2.0" },
+]
 obs = [
     { name = "opentelemetry-exporter-otlp", specifier = ">=1.24" },
     { name = "opentelemetry-sdk", specifier = ">=1.24" },
@@ -827,6 +868,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/02/08/9c41fb51ab5b43eb21674aff13df270e8ba6c4b29c8624e328dc7a9482af/distlib-0.4.3-py2.py3-none-any.whl", hash = "sha256:4b0ce306c966eb73bc3a7b6abad017c556dadd92c44701562cd528ac7fde4d5b", size = 470628, upload-time = "2026-06-12T08:04:50.506Z" },
 ]
 
+[[package]]
+name = "dnspython"
+version = "2.8.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/8c/8b/57666417c0f90f08bcafa776861060426765fdb422eb10212086fb811d26/dnspython-2.8.0.tar.gz", hash = "sha256:181d3c6996452cb1189c4046c61599b84a5a86e099562ffde77d26984ff26d0f", size = 368251, upload-time = "2025-09-07T18:58:00.022Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/ba/5a/18ad964b0086c6e62e2e7500f7edc89e3faa45033c71c1893d34eed2b2de/dnspython-2.8.0-py3-none-any.whl", hash = "sha256:01d9bbc4a2d76bf0db7c1f729812ded6d912bd318d3b1cf81d30c0f845dbf3af", size = 331094, upload-time = "2025-09-07T18:57:58.071Z" },
+]
+
 [[package]]
 name = "docker"
 version = "7.2.0"
@@ -841,6 +891,19 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/75/23/529140fe1aab80fc6992f93a706deec709140a6397439139a054e1515c45/docker-7.2.0-py3-none-any.whl", hash = "sha256:a3f45fdeb9165e2d25d9a1d02ddf3bc70fb572cf5ebbf9b58558c22caf29b71f", size = 148775, upload-time = "2026-07-09T14:53:45.224Z" },
 ]
 
+[[package]]
+name = "email-validator"
+version = "2.3.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "dnspython" },
+    { name = "idna" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/f5/22/900cb125c76b7aaa450ce02fd727f452243f2e91a61af068b40adba60ea9/email_validator-2.3.0.tar.gz", hash = "sha256:9fc05c37f2f6cf439ff414f8fc46d917929974a82244c20eb10231ba60c54426", size = 51238, upload-time = "2025-08-26T13:09:06.831Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/de/15/545e2b6cf2e3be84bc1ed85613edd75b8aea69807a71c26f4ca6a9258e82/email_validator-2.3.0-py3-none-any.whl", hash = "sha256:80f13f623413e6b197ae73bb10bf4eb0908faf509ad8362c5edeb0be7fd450b4", size = 35604, upload-time = "2025-08-26T13:09:05.858Z" },
+]
+
 [[package]]
 name = "fastapi"
 version = "0.141.1"