M4: admin manual-review queue and operator actions (#017, #018)

- ReviewQueueRepository: filtered/paginated queue query + item loader.
- Admin panel: /admin/review list/detail with extracted-text preview.
- Operator actions: send-to-analysis, complete, reject with Document Slot
  compensation and row-level status guard.
- /code-review cleanups applied; actions lock via DocumentRepository.

Closes issues 017, 018.

src/contract_check/api/admin/review.py

@@ -0,0 +1,250 @@
+"""Manual-review queue routes for the admin panel.
+
+List / detail / three operator actions (send to analysis, complete, reject).
+All mutations are HTMX-driven and protected by the ``HX-Request`` CSRF guard.
+"""
+
+from __future__ import annotations
+
+import datetime as dt
+import json
+import uuid
+from decimal import Decimal
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, Form, HTTPException, Request, status
+from fastapi.responses import HTMLResponse, RedirectResponse
+
+from src.contract_check.api.admin.auth import AdminUser, HtmxGuard, require_admin
+from src.contract_check.api.admin.templating import templates
+from src.contract_check.api.deps import AsyncSessionDep, PublisherDep
+from src.contract_check.core.config import get_settings
+from src.contract_check.core.db.repositories import ReviewQueueRepository
+from src.contract_check.core.logging import get_logger
+from src.contract_check.core.review.actions import (
+    ReviewActionError,
+    complete,
+    reject,
+    send_to_analysis,
+)
+from src.contract_check.core.s3.minio_storage import MinioStorage, MinioStorageError
+
+log = get_logger(__name__)
+
+router = APIRouter(
+    prefix="/admin/review",
+    tags=["admin-review"],
+    dependencies=[Depends(require_admin)],
+)
+
+PAGE_SIZE = 25
+
+
+def _storage() -> MinioStorage:
+    settings = get_settings()
+    return MinioStorage.from_endpoint_url(
+        endpoint_url=settings.s3_endpoint_url,
+        access_key=settings.s3_access_key,
+        secret_key=settings.s3_secret_key,
+        bucket=settings.s3_bucket,
+        region=settings.s3_region,
+    )
+
+
+async def _text_preview(extracted_s3_key: str | None) -> str | None:
+    if not extracted_s3_key:
+        return None
+    try:
+        text_bytes = await _storage().get(extracted_s3_key)
+        text = text_bytes.decode("utf-8", errors="replace")
+        return text[:4000] + ("\n…" if len(text) > 4000 else "")
+    except MinioStorageError:
+        return None
+
+
+def _date_or_none(value: str) -> dt.date | None:
+    try:
+        return dt.date.fromisoformat(value)
+    except ValueError:
+        return None
+
+
+def _decimal_or_none(value: str) -> Decimal | None:
+    value = value.strip()
+    if not value:
+        return None
+    try:
+        return Decimal(value)
+    except Exception:
+        return None
+
+
+@router.get("", response_class=HTMLResponse, include_in_schema=False)
+async def list_review_queue(
+    request: Request,
+    session: AsyncSessionDep,
+    date_from: str = "",
+    date_to: str = "",
+    conf_min: str = "",
+    conf_max: str = "",
+    page: int = 1,
+) -> HTMLResponse:
+    """Paginated manual-review backlog with date/confidence filters."""
+    page = max(page, 1)
+    offset = (page - 1) * PAGE_SIZE
+
+    date_from_parsed = _date_or_none(date_from)
+    date_to_parsed = _date_or_none(date_to)
+    conf_min_parsed = _decimal_or_none(conf_min)
+    conf_max_parsed = _decimal_or_none(conf_max)
+
+    repo = ReviewQueueRepository(session)
+    rows, total = await repo.list_queue(
+        limit=PAGE_SIZE,
+        offset=offset,
+        date_from=date_from_parsed,
+        date_to=date_to_parsed,
+        conf_min=conf_min_parsed,
+        conf_max=conf_max_parsed,
+    )
+    pages = max(1, (total + PAGE_SIZE - 1) // PAGE_SIZE)
+
+    response = templates.TemplateResponse(
+        request,
+        "review_list.html",
+        {
+            "request": request,
+            "items": rows,
+            "total": total,
+            "page": page,
+            "pages": pages,
+            "date_from": date_from,
+            "date_to": date_to,
+            "conf_min": conf_min,
+            "conf_max": conf_max,
+        },
+    )
+    _attach_toast(response, request.query_params.get("toast"))
+    return response
+
+
+@router.get("/{document_id}", response_class=HTMLResponse, include_in_schema=False)
+async def review_detail(
+    request: Request,
+    session: AsyncSessionDep,
+    document_id: uuid.UUID,
+) -> HTMLResponse:
+    """Detail page: prescreen metadata + extracted text preview + actions."""
+    item = await ReviewQueueRepository(session).get_item(document_id)
+    if item is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="document not found")
+
+    preview = await _text_preview(item.extracted_s3_key)
+    response = templates.TemplateResponse(
+        request,
+        "review_detail.html",
+        {
+            "request": request,
+            "item": item,
+            "preview": preview,
+            "toast": request.query_params.get("toast"),
+        },
+    )
+    _attach_toast(response, request.query_params.get("toast"))
+    return response
+
+
+def _attach_toast(response: HTMLResponse, toast: str | None) -> None:
+    if toast:
+        response.headers["HX-Trigger"] = json.dumps({"showToast": toast})
+
+
+def _review_action_redirect(
+    *,
+    document_id: uuid.UUID,
+    success_toast: str,
+    error: ReviewActionError | None = None,
+) -> RedirectResponse:
+    if error is not None:
+        target = f"/admin/review/{document_id}?toast={error.message}"
+    else:
+        target = f"/admin/review?toast={success_toast}"
+    return RedirectResponse(url=target, status_code=status.HTTP_303_SEE_OTHER)
+
+
+@router.post(
+    "/{document_id}/send-to-analysis", response_class=HTMLResponse, include_in_schema=False
+)
+async def send_to_analysis_action(
+    request: Request,
+    session: AsyncSessionDep,
+    publisher: PublisherDep,
+    document_id: uuid.UUID,
+    admin: Annotated[AdminUser, Depends(require_admin)],
+    _: Annotated[None, HtmxGuard],
+) -> RedirectResponse:
+    try:
+        await send_to_analysis(session, publisher, document_id=document_id)
+    except ReviewActionError as exc:
+        return _review_action_redirect(document_id=document_id, success_toast="", error=exc)
+    log.info(
+        "admin_review_action",
+        action="send_to_analysis",
+        document_id=str(document_id),
+        admin_user_id=str(admin.user_id),
+    )
+    return _review_action_redirect(
+        document_id=document_id,
+        success_toast="Документ отправлен на полный анализ",
+    )
+
+
+@router.post("/{document_id}/complete", response_class=HTMLResponse, include_in_schema=False)
+async def complete_action(
+    request: Request,
+    session: AsyncSessionDep,
+    document_id: uuid.UUID,
+    admin: Annotated[AdminUser, Depends(require_admin)],
+    _: Annotated[None, HtmxGuard],
+    note: Annotated[str, Form()] = "",
+) -> RedirectResponse:
+    try:
+        await complete(session, document_id=document_id, note=note)
+    except ReviewActionError as exc:
+        return _review_action_redirect(document_id=document_id, success_toast="", error=exc)
+    log.info(
+        "admin_review_action",
+        action="complete",
+        document_id=str(document_id),
+        admin_user_id=str(admin.user_id),
+    )
+    return _review_action_redirect(
+        document_id=document_id,
+        success_toast="Документ закрыт с пометкой оператора",
+    )
+
+
+@router.post("/{document_id}/reject", response_class=HTMLResponse, include_in_schema=False)
+async def reject_action(
+    request: Request,
+    session: AsyncSessionDep,
+    document_id: uuid.UUID,
+    admin: Annotated[AdminUser, Depends(require_admin)],
+    _: Annotated[None, HtmxGuard],
+    reason: Annotated[str, Form()] = "",
+) -> RedirectResponse:
+    try:
+        await reject(session, document_id=document_id, reason=reason)
+    except ReviewActionError as exc:
+        return _review_action_redirect(document_id=document_id, success_toast="", error=exc)
+    log.info(
+        "admin_review_action",
+        action="reject",
+        document_id=str(document_id),
+        admin_user_id=str(admin.user_id),
+        reason=reason,
+    )
+    return _review_action_redirect(
+        document_id=document_id,
+        success_toast="Документ отклонён, Document Slot возвращён",
+    )

src/contract_check/api/admin/router.py

@@ -14,6 +14,7 @@ from fastapi.responses import HTMLResponse, RedirectResponse
 
 from src.contract_check.api.admin.auth import ADMIN_COOKIE, ADMIN_COOKIE_MAX_AGE, resolve_admin
 from src.contract_check.api.admin.billing import router as billing_router
+from src.contract_check.api.admin.review import router as review_router
 from src.contract_check.api.admin.templating import templates
 from src.contract_check.api.admin.users import router as users_router
 from src.contract_check.api.deps import AsyncSessionDep, RateLimiterDep, require_auth_rate_limit
@@ -28,6 +29,7 @@ log = get_logger(__name__)
 router = APIRouter(tags=["admin"])
 router.include_router(users_router)
 router.include_router(billing_router)
+router.include_router(review_router)
 
 
 @router.get("/admin", include_in_schema=False)

src/contract_check/api/admin/templates/base.html

@@ -70,6 +70,7 @@
     <nav>
       <a href="/admin/users" class="{{ 'active' if request.url.path.startswith('/admin/users') else '' }}">Пользователи</a>
       <a href="/admin/invoices" class="{{ 'active' if request.url.path.startswith('/admin/invoices') else '' }}">Счета</a>
+      <a href="/admin/review" class="{{ 'active' if request.url.path.startswith('/admin/review') else '' }}">Ручная проверка</a>
     </nav>
     <span class="spacer"></span>
     <form method="post" action="/admin/logout"><button class="btn ghost sm" type="submit">Выйти</button></form>

src/contract_check/api/admin/templates/review_detail.html

@@ -0,0 +1,72 @@
+{% extends "base.html" %}
+{% block title %}Ручная проверка — {{ item.filename }}{% endblock %}
+{% block content %}
+<div class="toolbar">
+  <a class="btn ghost sm" href="/admin/review">← Назад к очереди</a>
+</div>
+
+<div class="grid cols-2">
+  <div class="card">
+    <h2 style="margin-top:0">{{ item.filename }}</h2>
+    <p class="muted">ID: <code>{{ item.document_id }}</code></p>
+
+    <h3 style="margin-top:18px">Реквизиты договора</h3>
+    <table>
+      <tbody>
+        <tr><td class="muted">Тип</td><td>{{ item.contract_type or '—' }}</td></tr>
+        <tr><td class="muted">Сторона А</td><td>{{ item.party_a or '—' }}</td></tr>
+        <tr><td class="muted">Сторона Б</td><td>{{ item.party_b or '—' }}</td></tr>
+        <tr><td class="muted">Сумма</td><td>{% if item.total_amount %}{{ item.total_amount }} {{ item.currency or '' }}{% else %}—{% endif %}</td></tr>
+        <tr><td class="muted">Срок</td><td>{{ item.start_date or '—' }} — {{ item.end_date or '—' }}</td></tr>
+        <tr><td class="muted">Неустойка</td><td>{{ item.has_penalty_clause | yn }}</td></tr>
+        <tr><td class="muted">Расторжение</td><td>{{ item.has_termination_clause | yn }}</td></tr>
+        <tr><td class="muted">Арбитраж</td><td>{{ item.has_arbitration | yn }}</td></tr>
+        <tr><td class="muted">Уверенность</td><td><span class="pill">{{ (item.confidence_score * 100)|round(1) }}%</span></td></tr>
+        <tr><td class="muted">Пресскрин</td><td>{{ item.prescreened_at | dt }}</td></tr>
+      </tbody>
+    </table>
+
+    {% if item.auto_findings %}
+    <h3 style="margin-top:18px">Находки</h3>
+    <ul>
+      {% for f in item.auto_findings %}
+      <li><code>{{ f }}</code></li>
+      {% endfor %}
+    </ul>
+    {% endif %}
+
+    {% if item.auto_summary %}
+    <h3 style="margin-top:18px">Авто-сводка</h3>
+    <p class="muted">{{ item.auto_summary }}</p>
+    {% endif %}
+  </div>
+
+  <div class="card">
+    <h3 style="margin-top:0">Превью текста</h3>
+    {% if preview %}
+    <pre style="white-space:pre-wrap;word-break:break-word;background:#0f1115;padding:12px;border-radius:8px;max-height:300px;overflow:auto">{{ preview }}</pre>
+    {% else %}
+    <p class="muted">Превью недоступно.</p>
+    {% endif %}
+
+    <h3 style="margin-top:18px">Действия</h3>
+
+    <form method="post" action="/admin/review/{{ item.document_id }}/send-to-analysis" class="card" style="margin:12px 0 0">
+      <p class="muted">Отправить документ на полный анализ ИИ.</p>
+      <button class="btn" type="submit" hx-post="/admin/review/{{ item.document_id }}/send-to-analysis" hx-confirm="Отправить на анализ?">Отправить на анализ</button>
+    </form>
+
+    <form method="post" action="/admin/review/{{ item.document_id }}/complete" class="card" style="margin:12px 0 0">
+      <label for="note">Пометка оператора</label>
+      <textarea id="note" name="note" rows="3" placeholder="Причина закрытия, видимая клиенту..."></textarea>
+      <button class="btn" type="submit" hx-post="/admin/review/{{ item.document_id }}/complete" style="margin-top:10px">Закрыть как завершённое</button>
+    </form>
+
+    <form method="post" action="/admin/review/{{ item.document_id }}/reject" class="card" style="margin:12px 0 0;border-color:var(--bad)">
+      <label for="reason">Причина отклонения</label>
+      <input id="reason" name="reason" placeholder="Не подлежит автоматической проверке...">
+      <button class="btn danger" type="submit" hx-post="/admin/review/{{ item.document_id }}/reject" hx-confirm="Отклонить и вернуть Document Slot?" style="margin-top:10px">Отклонить и вернуть</button>
+    </form>
+  </div>
+</div>
+{% endblock %}

src/contract_check/api/admin/templates/review_list.html

@@ -0,0 +1,59 @@
+{% extends "base.html" %}
+{% block title %}Ручная проверка — Админка{% endblock %}
+{% block content %}
+<div class="card">
+  <div class="toolbar">
+    <h2 style="margin:0">Ручная проверка <span class="muted" style="font-weight:400;font-size:14px">({{ total }})</span></h2>
+    <span class="spacer"></span>
+    <form method="get" action="/admin/review" class="row">
+      <input type="date" name="date_from" value="{{ date_from }}" placeholder="с">
+      <input type="date" name="date_to" value="{{ date_to }}" placeholder="по">
+      <input type="number" name="conf_min" value="{{ conf_min }}" min="0" max="1" step="0.01" placeholder="conf min" style="width:90px">
+      <input type="number" name="conf_max" value="{{ conf_max }}" min="0" max="1" step="0.01" placeholder="conf max" style="width:90px">
+      <button class="btn ghost sm" type="submit">Фильтр</button>
+    </form>
+  </div>
+
+  <table>
+    <thead>
+      <tr>
+        <th>Документ</th>
+        <th>Пользователь</th>
+        <th>Стороны</th>
+        <th>Сумма</th>
+        <th>Уверенность</th>
+        <th>Находки</th>
+        <th>Создан</th>
+        <th></th>
+      </tr>
+    </thead>
+    <tbody>
+      {% for item in items %}
+      <tr>
+        <td>{{ item.filename }}</td>
+        <td>
+          <a href="/admin/users/{{ item.user_id }}">{{ item.email or 'tg ' ~ item.telegram_id }}</a>
+        </td>
+        <td class="muted">{{ item.party_a or '—' }} → {{ item.party_b or '—' }}</td>
+        <td class="muted">{% if item.total_amount %}{{ item.total_amount }} {{ item.currency or '' }}{% else %}—{% endif %}</td>
+        <td><span class="pill">{{ (item.confidence_score * 100)|round(1) }}%</span></td>
+        <td class="muted">{{ item.auto_findings | length }}</td>
+        <td class="muted">{{ item.created_at | dt }}</td>
+        <td><a class="btn ghost sm" href="/admin/review/{{ item.document_id }}">Открыть</a></td>
+      </tr>
+      {% else %}
+      <tr><td colspan="8"><div class="empty">Очередь пуста</div></td></tr>
+      {% endfor %}
+    </tbody>
+  </table>
+
+  {% if pages > 1 %}
+  <div class="pager">
+    {% for p in range(1, pages + 1) %}
+      {% if p == page %}<span class="btn sm" style="background:var(--line);cursor:default">{{ p }}</span>
+      {% else %}<a class="btn ghost sm" href="/admin/review?date_from={{ date_from }}&date_to={{ date_to }}&conf_min={{ conf_min }}&conf_max={{ conf_max }}&page={{ p }}">{{ p }}</a>{% endif %}
+    {% endfor %}
+  </div>
+  {% endif %}
+</div>
+{% endblock %}

src/contract_check/api/admin/templating.py

@@ -25,4 +25,13 @@ def _format_dt(value: object, fmt: str = "%Y-%m-%d %H:%M UTC") -> str:
     return str(value)
 
 
+def _yes_no_none(value: object) -> str:
+    if value is True:
+        return "да"
+    if value is False:
+        return "нет"
+    return "—"
+
+
 templates.env.filters["dt"] = _format_dt
+templates.env.filters["yn"] = _yes_no_none

src/contract_check/core/db/repositories/__init__.py

@@ -19,6 +19,7 @@ from src.contract_check.core.db.repositories.passkeys import PasskeyRepository
 from src.contract_check.core.db.repositories.prescreen_results import PrescreenResultRepository
 from src.contract_check.core.db.repositories.profiles import UserProfileRepository
 from src.contract_check.core.db.repositories.reports import ReportRepository
+from src.contract_check.core.db.repositories.review import ReviewQueueRepository
 from src.contract_check.core.db.repositories.service_tokens import ServiceTokenRepository
 from src.contract_check.core.db.repositories.subscriptions import SubscriptionsRepository
 from src.contract_check.core.db.repositories.users import UserRepository
@@ -33,6 +34,7 @@ __all__ = [
     "PasskeyRepository",
     "PrescreenResultRepository",
     "ReportRepository",
+    "ReviewQueueRepository",
     "ServiceTokenRepository",
     "SubscriptionsRepository",
     "UserProfileRepository",

src/contract_check/core/db/repositories/review.py

@@ -0,0 +1,188 @@
+"""Manual-review queue data access.
+
+Joins documents in ``manual_review`` with their latest prescreen result and
+owning user so an operator can triage them without leaving the admin panel.
+"""
+
+from __future__ import annotations
+
+import datetime as dt
+import json
+import uuid
+from dataclasses import dataclass
+from decimal import Decimal
+from typing import Any
+
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+
+@dataclass(frozen=True, slots=True)
+class ReviewQueueItem:
+    """One manual-review backlog row with prescreen metadata and user identity."""
+
+    document_id: uuid.UUID
+    user_id: uuid.UUID
+    email: str | None
+    telegram_id: int | None
+    filename: str
+    status: str
+    created_at: dt.datetime
+    extracted_s3_key: str | None
+    prescreen_result_id: uuid.UUID
+    correlation_id: uuid.UUID
+    contract_type: str | None
+    party_a: str | None
+    party_b: str | None
+    total_amount: Decimal | None
+    currency: str | None
+    start_date: dt.date | None
+    end_date: dt.date | None
+    has_penalty_clause: bool | None
+    has_termination_clause: bool | None
+    has_arbitration: bool | None
+    confidence_score: Decimal | None
+    auto_summary: str | None
+    auto_findings: list[dict[str, Any]]
+    prescreened_at: dt.datetime
+
+
+class ReviewQueueRepository:
+    """Query the manual-review backlog and load one item for the detail page."""
+
+    _SELECT_COLUMNS = (
+        "d.id, d.user_id, u.email, u.telegram_id, d.filename, d.status, d.created_at, "
+        "d.extracted_s3_key, "
+        "pr.id AS prescreen_result_id, pr.correlation_id, pr.contract_type, pr.party_a, "
+        "pr.party_b, pr.total_amount, pr.currency, pr.start_date, pr.end_date, "
+        "pr.has_penalty_clause, pr.has_termination_clause, pr.has_arbitration, "
+        "pr.confidence_score, pr.auto_summary, pr.auto_findings, pr.prescreened_at"
+    )
+
+    def __init__(self, session: AsyncSession) -> None:
+        self._session = session
+
+    async def list_queue(
+        self,
+        *,
+        limit: int,
+        offset: int,
+        date_from: dt.date | None = None,
+        date_to: dt.date | None = None,
+        conf_min: Decimal | float | None = None,
+        conf_max: Decimal | float | None = None,
+    ) -> tuple[list[ReviewQueueItem], int]:
+        """Return paginated backlog rows plus total count.
+
+        Ordering is newest-first by document creation time. Confidence and date
+        filters apply to the selected (latest) prescreen result per document.
+        """
+        params: dict[str, Any] = {"limit": limit, "offset": offset}
+        date_clauses = ["d.status = 'manual_review'"]
+        if date_from is not None:
+            date_clauses.append("CAST(d.created_at AS DATE) >= :date_from")
+            params["date_from"] = date_from
+        if date_to is not None:
+            date_clauses.append("CAST(d.created_at AS DATE) <= :date_to")
+            params["date_to"] = date_to
+
+        where = " AND ".join(date_clauses)
+
+        # Inner query picks the latest prescreen result per document.
+        sql = (
+            f"SELECT * FROM ("
+            f"  SELECT DISTINCT ON (d.id) {self._SELECT_COLUMNS}"
+            f"  FROM documents d"
+            f"  JOIN users u ON u.id = d.user_id"
+            f"  JOIN prescreen_results pr ON pr.document_id = d.id"
+            f"  WHERE {where}"
+            f"  ORDER BY d.id, pr.prescreened_at DESC"
+            f") q"
+        )
+
+        outer_clauses: list[str] = []
+        if conf_min is not None:
+            outer_clauses.append("q.confidence_score >= :conf_min")
+            params["conf_min"] = float(conf_min)
+        if conf_max is not None:
+            outer_clauses.append("q.confidence_score <= :conf_max")
+            params["conf_max"] = float(conf_max)
+
+        outer_where = "1 = 1"
+        if outer_clauses:
+            outer_where = " AND ".join(outer_clauses)
+
+        rows_result = await self._session.execute(
+            text(
+                f"{sql} WHERE {outer_where} ORDER BY q.created_at DESC LIMIT :limit OFFSET :offset"
+            ),
+            params,
+        )
+        rows = [self._row_to_item(row) for row in rows_result.all()]
+
+        count_result = await self._session.execute(
+            text(
+                f"SELECT count(DISTINCT d.id) FROM documents d "
+                f"JOIN prescreen_results pr ON pr.document_id = d.id "
+                f"WHERE {where}"
+            ),
+            {k: v for k, v in params.items() if k not in ("limit", "offset")},
+        )
+        total = int(count_result.scalar_one())
+        return rows, total
+
+    async def get_item(self, document_id: uuid.UUID) -> ReviewQueueItem | None:
+        """Load a single backlog row by document id, or None if not manual-review."""
+        result = await self._session.execute(
+            text(
+                f"SELECT {self._SELECT_COLUMNS} "
+                f"FROM documents d "
+                f"JOIN users u ON u.id = d.user_id "
+                f"JOIN prescreen_results pr ON pr.document_id = d.id "
+                f"WHERE d.id = :d AND d.status = 'manual_review' "
+                f"ORDER BY pr.prescreened_at DESC "
+                f"LIMIT 1"
+            ),
+            {"d": document_id},
+        )
+        row = result.first()
+        if row is None:
+            return None
+        return self._row_to_item(row)
+
+    @staticmethod
+    def _row_to_item(row: Any) -> ReviewQueueItem:
+        raw_findings = row.auto_findings
+        findings: list[dict[str, Any]]
+        if isinstance(raw_findings, str):
+            findings = json.loads(raw_findings)
+        elif raw_findings is None:
+            findings = []
+        else:
+            findings = list(raw_findings)
+        return ReviewQueueItem(
+            document_id=row.id,
+            user_id=row.user_id,
+            email=row.email,
+            telegram_id=row.telegram_id,
+            filename=row.filename,
+            status=row.status,
+            created_at=row.created_at,
+            extracted_s3_key=row.extracted_s3_key,
+            prescreen_result_id=row.prescreen_result_id,
+            correlation_id=row.correlation_id,
+            contract_type=row.contract_type,
+            party_a=row.party_a,
+            party_b=row.party_b,
+            total_amount=row.total_amount,
+            currency=row.currency,
+            start_date=row.start_date,
+            end_date=row.end_date,
+            has_penalty_clause=row.has_penalty_clause,
+            has_termination_clause=row.has_termination_clause,
+            has_arbitration=row.has_arbitration,
+            confidence_score=row.confidence_score,
+            auto_summary=row.auto_summary,
+            auto_findings=findings,
+            prescreened_at=row.prescreened_at,
+        )

src/contract_check/core/review/__init__.py

@@ -0,0 +1,3 @@
+"""Manual-review queue operator actions (send/complete/reject)."""
+
+from __future__ import annotations

src/contract_check/core/review/actions.py

@@ -0,0 +1,270 @@
+"""Manual-review queue operator actions.
+
+Three resolvers for a document stuck in ``manual_review``:
+
+- ``send_to_analysis`` returns the document to the analyze pipeline.
+- ``complete`` closes it as a lightweight admin-reviewed report.
+- ``reject`` terminates it as failed and compensates the consumed Document Slot
+  unconditionally (admin judgment).
+
+All actions lock the document row, require the current status to be
+``manual_review``, and commit their own changes. Publishing happens after the
+status change so the analyze worker never sees the message while the document
+is still terminal.
+"""
+
+from __future__ import annotations
+
+import datetime as dt
+import uuid
+
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.contract_check.core.billing.quota import compensate_document_slot
+from src.contract_check.core.db.repositories import (
+    DocumentRepository,
+    JobRepository,
+    ReportRepository,
+    ReviewQueueRepository,
+)
+from src.contract_check.core.db.repositories.review import ReviewQueueItem
+from src.contract_check.core.logging import get_logger
+from src.contract_check.core.mq.messages import AnalyzeRequested
+from src.contract_check.core.mq.publisher import Publisher
+from src.contract_check.core.mq.topology import RK_ANALYZE
+
+log = get_logger(__name__)
+
+
+class ReviewActionError(Exception):
+    """User-facing error returned as a toast in the admin panel."""
+
+    def __init__(self, message: str) -> None:
+        self.message = message
+        super().__init__(message)
+
+
+async def send_to_analysis(
+    session: AsyncSession,
+    publisher: Publisher,
+    *,
+    document_id: uuid.UUID,
+) -> None:
+    """Move a manual-review document back to the analyze pipeline.
+
+    Status is changed to ``analyzing`` before publishing so a racing analyze
+    worker does not treat the message as a terminal redelivery. If publishing
+    fails, the status is rolled back to ``manual_review``.
+    """
+    item = await ReviewQueueRepository(session).get_item(document_id)
+    if item is None:
+        raise ReviewActionError("Документ не найден в очереди ручной проверки.")
+
+    if not item.extracted_s3_key:
+        raise ReviewActionError("У документа отсутствует извлечённый текст — отправка невозможна.")
+
+    status = await _require_manual_review_for_update(session, document_id)
+    if status != "manual_review":
+        raise ReviewActionError("Документ уже не в статусе ручной проверки.")
+
+    await _set_status(session, document_id, status="analyzing", stage="queued_analyze")
+    await JobRepository(session).upsert_running(
+        document_id,
+        item.correlation_id,
+        queue="analyze",
+    )
+    await session.commit()
+
+    prescreen_meta = _prescreen_meta_from_item(item)
+    next_msg = AnalyzeRequested(
+        correlation_id=item.correlation_id,
+        document_id=item.document_id,
+        user_id=item.user_id,
+        extracted_s3_key=item.extracted_s3_key,
+        filename=item.filename,
+        char_count=0,
+        ocr_used=False,
+        is_structured=False,
+        has_tables=False,
+        prescreen_meta=prescreen_meta,
+        prescreen_result_id=item.prescreen_result_id,
+        attempt=0,
+    )
+
+    try:
+        await publisher.publish(next_msg, routing_key=RK_ANALYZE)
+    except Exception:
+        await _set_status(session, document_id, status="manual_review", stage="manual_review")
+        await session.commit()
+        raise
+
+    log.info(
+        "admin_review_action",
+        action="send_to_analysis",
+        document_id=str(document_id),
+        user_id=str(item.user_id),
+        prescreen_result_id=str(item.prescreen_result_id),
+        correlation_id=str(item.correlation_id),
+    )
+
+
+async def complete(
+    session: AsyncSession,
+    *,
+    document_id: uuid.UUID,
+    note: str,
+) -> None:
+    """Close a manual-review document as ``done`` with an operator note.
+
+    A lightweight report is written so the customer sees a result; no LLM
+    is invoked.
+    """
+    item = await ReviewQueueRepository(session).get_item(document_id)
+    if item is None:
+        raise ReviewActionError("Документ не найден в очереди ручной проверки.")
+
+    status = await _require_manual_review_for_update(session, document_id)
+    if status != "manual_review":
+        raise ReviewActionError("Документ уже не в статусе ручной проверки.")
+
+    safe_note = (note or "Закрыто оператором без полного анализа.").strip()
+    content_json: dict[str, object] = {
+        "findings": [
+            {
+                "title": "Проверка завершена оператором",
+                "severity": "low",
+                "explanation": safe_note,
+                "recommendation": "",
+                "citations": [],
+            }
+        ]
+    }
+    markdown = _operator_note_markdown(item, safe_note)
+    await ReportRepository(session).upsert_analyze_report(
+        document_id=document_id,
+        content_json=content_json,
+        markdown=markdown,
+        model_used="admin-complete",
+        prompt_tokens=0,
+        eval_tokens=0,
+        latency_ms=0,
+        prescreen_meta=_prescreen_meta_from_item(item),
+        prescreen_result_id=item.prescreen_result_id,
+    )
+    await _set_status(session, document_id, status="done", stage="review_completed")
+    await session.commit()
+
+    log.info(
+        "admin_review_action",
+        action="complete",
+        document_id=str(document_id),
+        user_id=str(item.user_id),
+        prescreen_result_id=str(item.prescreen_result_id),
+    )
+
+
+async def reject(
+    session: AsyncSession,
+    *,
+    document_id: uuid.UUID,
+    reason: str,
+) -> None:
+    """Reject a manual-review document and compensate the Document Slot.
+
+    Compensation is unconditional (``policy='all'``) because this is an
+    operator judgment call, not an automated failure classification.
+    """
+    item = await ReviewQueueRepository(session).get_item(document_id)
+    if item is None:
+        raise ReviewActionError("Документ не найден в очереди ручной проверки.")
+
+    status = await _require_manual_review_for_update(session, document_id)
+    if status != "manual_review":
+        raise ReviewActionError("Документ уже не в статусе ручной проверки.")
+
+    await _set_status(session, document_id, status="failed", stage="review_rejected")
+    await compensate_document_slot(
+        session,
+        document_id,
+        "manual_review_rejected",
+        "all",
+    )
+    await session.commit()
+
+    log.info(
+        "admin_review_action",
+        action="reject",
+        document_id=str(document_id),
+        user_id=str(item.user_id),
+        reason=reason or "",
+    )
+
+
+async def _require_manual_review_for_update(session: AsyncSession, document_id: uuid.UUID) -> str:
+    status = await DocumentRepository(session).get_status_for_update(document_id)
+    if status is None:
+        raise ReviewActionError("Документ не найден.")
+    return status
+
+
+async def _set_status(
+    session: AsyncSession,
+    document_id: uuid.UUID,
+    *,
+    status: str,
+    stage: str,
+) -> None:
+    await session.execute(
+        text("UPDATE documents SET status = :s, stage = :st WHERE id = :d"),
+        {"s": status, "st": stage, "d": document_id},
+    )
+
+
+def _prescreen_meta_from_item(item: ReviewQueueItem) -> dict[str, object]:
+    def _iso(d: dt.date | None) -> str | None:
+        return d.isoformat() if d else None
+
+    return {
+        "correlation_id": str(item.correlation_id),
+        "document_id": str(item.document_id),
+        "user_id": str(item.user_id),
+        "attempt": 0,
+        "text_s3_key": item.extracted_s3_key,
+        "filename": item.filename,
+        "prescreened_at": item.prescreened_at.isoformat(),
+        "contract_type": item.contract_type,
+        "party_a": item.party_a,
+        "party_b": item.party_b,
+        "total_amount": float(item.total_amount) if item.total_amount is not None else None,
+        "currency": item.currency,
+        "start_date": _iso(item.start_date),
+        "end_date": _iso(item.end_date),
+        "has_penalty_clause": item.has_penalty_clause,
+        "has_termination_clause": item.has_termination_clause,
+        "has_arbitration": item.has_arbitration,
+        "confidence_score": float(item.confidence_score)
+        if item.confidence_score is not None
+        else 0.0,
+        "routing_decision": "manual_review",
+        "auto_summary": item.auto_summary,
+        "auto_findings": item.auto_findings,
+    }
+
+
+def _operator_note_markdown(item: ReviewQueueItem, note: str) -> str:
+    lines: list[str] = [
+        f"# Проверка договора: {item.contract_type or 'договор'}",
+        "",
+        note,
+        "",
+        "**Решение:** проверка завершена оператором без полного анализа ИИ.",
+    ]
+    if item.party_a:
+        lines.append(f"- **Сторона А:** {item.party_a}")
+    if item.party_b:
+        lines.append(f"- **Сторона Б:** {item.party_b}")
+    if item.total_amount is not None:
+        lines.append(f"- **Сумма:** {item.total_amount} {item.currency or ''}")
+    lines.extend(["", "---", "", "*Это операторская пометка, а не полный автоматический отчёт.*"])
+    return "\n".join(lines)

tests/integration/test_admin_review.py

@@ -0,0 +1,335 @@
+"""Admin manual-review queue HTTP seam tests (issue 017 + 018).
+
+Drives the server-rendered admin panel through the ASGI transport with real
+Postgres / RabbitMQ / MinIO. Covers list/detail/filters, RBAC, CSRF guard, and
+all three review actions.
+"""
+
+from __future__ import annotations
+
+import datetime as dt
+import uuid
+from decimal import Decimal
+from typing import Any
+
+import aio_pika
+import httpx
+import pytest
+from sqlalchemy import text
+
+from contract_check.core.db.session import create_session_factory
+from contract_check.core.mq.messages import AnalyzeRequested
+from contract_check.core.mq.topology import QUEUE_ANALYZE
+from contract_check.core.s3 import extracted_key
+from contract_check.core.s3.minio_storage import MinioStorage
+from contract_check.core.security.passwords import hash_password
+
+pytestmark = pytest.mark.integration
+
+
+_TEST_TEXT = "Договор поставки. " * 50
+
+
+def _storage(infra: dict[str, str]) -> MinioStorage:
+    return MinioStorage.from_endpoint_url(
+        endpoint_url=infra["s3_endpoint_url"],
+        access_key=infra["s3_access_key"],
+        secret_key=infra["s3_secret_key"],
+        bucket="contract-check-docs",
+    )
+
+
+def _session_factory() -> Any:
+    return create_session_factory()
+
+
+async def _seed_admin(db_session: Any, *, email: str, password: str) -> str:
+    result = await db_session.execute(
+        text(
+            "INSERT INTO users (email, password_hash, role, is_active) "
+            "VALUES (:e, :p, 'admin', TRUE) RETURNING id"
+        ),
+        {"e": email, "p": hash_password(password)},
+    )
+    await db_session.commit()
+    return str(result.scalar_one())
+
+
+async def _login_as(client: httpx.AsyncClient, email: str, password: str) -> None:
+    r = await client.post(
+        "/admin/login", data={"email": email, "password": password}, follow_redirects=False
+    )
+    assert r.status_code == 303
+    assert r.headers["location"].startswith("/admin/users")
+
+
+async def _seed_manual_review_doc(
+    infra: dict[str, str],
+    *,
+    telegram_id: int | None = None,
+    created_at: Any,
+    confidence: float = 0.5,
+) -> tuple[uuid.UUID, uuid.UUID]:
+    if telegram_id is None:
+        telegram_id = int(uuid.uuid4().int % 1_000_000_000)
+    sess = _session_factory()
+    store = _storage(infra)
+    async with sess() as session:
+        result = await session.execute(
+            text("INSERT INTO users (telegram_id, credits_left) VALUES (:t, 5) RETURNING id"),
+            {"t": telegram_id},
+        )
+        user_id = result.scalar_one()
+        await session.commit()
+
+    document_id = uuid.uuid4()
+    ext_key = extracted_key(str(user_id), str(document_id))
+    await store.put(ext_key, _TEST_TEXT.encode("utf-8"), content_type="text/plain; charset=utf-8")
+
+    async with sess() as session:
+        await session.execute(
+            text(
+                "INSERT INTO documents "
+                "(id, user_id, s3_key, extracted_s3_key, filename, mime, bytes, status, stage) "
+                "VALUES (:id, :uid, :s3, :ext, 'contract.pdf', 'application/pdf', :bytes, "
+                " 'manual_review', 'manual_review')"
+            ),
+            {
+                "id": document_id,
+                "uid": user_id,
+                "s3": f"users/{user_id}/docs/{document_id}.pdf",
+                "ext": ext_key,
+                "bytes": len(_TEST_TEXT),
+            },
+        )
+        await session.execute(
+            text(
+                "INSERT INTO prescreen_results "
+                "(id, document_id, correlation_id, contract_type, party_a, party_b, "
+                " total_amount, currency, start_date, end_date, has_penalty_clause, "
+                " has_termination_clause, has_arbitration, confidence_score, routing_decision, "
+                " prescreened_at, auto_summary, auto_findings, extractor_version) "
+                "VALUES (:id, :d, :cid, 'supply', 'ООО Продавец', 'ООО Покупатель', "
+                " :ta, 'RUB', '2025-01-01', '2025-12-31', TRUE, TRUE, FALSE, "
+                " :cs, 'manual_review', now(), 'summary', '[{\"note\":\"ok\"}]', 'heuristic-v2')"
+            ),
+            {
+                "id": uuid.uuid4(),
+                "d": document_id,
+                "cid": uuid.uuid4(),
+                "ta": Decimal("100000.00"),
+                "cs": confidence,
+            },
+        )
+        await session.execute(
+            text("UPDATE documents SET created_at = :ca WHERE id = :id"),
+            {"ca": created_at, "id": document_id},
+        )
+        await session.commit()
+    return user_id, document_id
+
+
+async def test_review_queue_redirects_without_session(client: httpx.AsyncClient) -> None:
+    r = await client.get("/admin/review", follow_redirects=False)
+    assert r.status_code == 303
+    assert r.headers["location"].startswith("/admin/login")
+
+
+async def test_review_queue_list_and_filters(
+    client: httpx.AsyncClient, db_session: Any, infra: dict[str, str]
+) -> None:
+    admin_email = f"admin-{uuid.uuid4().hex[:8]}@test.local"
+    await _seed_admin(db_session, email=admin_email, password="adminpass-123")
+    await _login_as(client, admin_email, "adminpass-123")
+
+    user_id, doc_id = await _seed_manual_review_doc(
+        infra,
+        created_at=dt.datetime(2030, 3, 15, tzinfo=dt.UTC),
+        confidence=0.8,
+    )
+    await _seed_manual_review_doc(
+        infra,
+        created_at=dt.datetime(2030, 6, 1, tzinfo=dt.UTC),
+        confidence=0.4,
+    )
+
+    # Unfiltered list in the unique date range.
+    r = await client.get(
+        "/admin/review?date_from=2030-01-01&date_to=2030-12-31", follow_redirects=False
+    )
+    assert r.status_code == 200
+    assert b"contract.pdf" in r.content
+    assert "ООО Продавец".encode() in r.content
+
+    # Date filter excludes the later document.
+    r = await client.get(
+        "/admin/review?date_from=2030-03-01&date_to=2030-03-31", follow_redirects=False
+    )
+    assert r.status_code == 200
+    assert str(doc_id).encode() in r.content
+
+    # Confidence filter.
+    r = await client.get(
+        "/admin/review?date_from=2030-01-01&date_to=2030-12-31&conf_min=0.7",
+        follow_redirects=False,
+    )
+    assert r.status_code == 200
+    assert b"80.0%" in r.content
+
+
+async def test_review_detail_with_preview(
+    client: httpx.AsyncClient, db_session: Any, infra: dict[str, str]
+) -> None:
+    admin_email = f"admin-{uuid.uuid4().hex[:8]}@test.local"
+    await _seed_admin(db_session, email=admin_email, password="adminpass-123")
+    await _login_as(client, admin_email, "adminpass-123")
+
+    user_id, doc_id = await _seed_manual_review_doc(
+        infra, created_at=dt.datetime(2030, 4, 1, tzinfo=dt.UTC)
+    )
+
+    r = await client.get(f"/admin/review/{doc_id}", follow_redirects=False)
+    assert r.status_code == 200
+    assert "ООО Продавец".encode() in r.content
+    assert "Договор поставки.".encode() in r.content
+    assert f"/admin/review/{doc_id}/send-to-analysis".encode() in r.content
+
+    # Non-existent document returns 404.
+    r = await client.get(f"/admin/review/{uuid.uuid4()}", follow_redirects=False)
+    assert r.status_code == 404
+
+
+async def test_review_action_requires_htmx_header(
+    client: httpx.AsyncClient, db_session: Any, infra: dict[str, str]
+) -> None:
+    admin_email = f"admin-{uuid.uuid4().hex[:8]}@test.local"
+    await _seed_admin(db_session, email=admin_email, password="adminpass-123")
+    await _login_as(client, admin_email, "adminpass-123")
+
+    _user_id, doc_id = await _seed_manual_review_doc(
+        infra, created_at=dt.datetime(2030, 5, 1, tzinfo=dt.UTC)
+    )
+
+    r = await client.post(
+        f"/admin/review/{doc_id}/complete",
+        data={"note": "note"},
+        follow_redirects=False,
+    )
+    assert r.status_code == 400
+
+
+async def test_review_complete_action(
+    client: httpx.AsyncClient, db_session: Any, infra: dict[str, str]
+) -> None:
+    admin_email = f"admin-{uuid.uuid4().hex[:8]}@test.local"
+    await _seed_admin(db_session, email=admin_email, password="adminpass-123")
+    await _login_as(client, admin_email, "adminpass-123")
+
+    _user_id, doc_id = await _seed_manual_review_doc(
+        infra, created_at=dt.datetime(2030, 6, 1, tzinfo=dt.UTC)
+    )
+
+    r = await client.post(
+        f"/admin/review/{doc_id}/complete",
+        data={"note": "Проверено оператором."},
+        headers={"HX-Request": "true"},
+        follow_redirects=False,
+    )
+    assert r.status_code == 303
+    assert "/admin/review" in r.headers["location"]
+
+    sess = _session_factory()
+    async with sess() as session:
+        status = (
+            await session.execute(text("SELECT status FROM documents WHERE id = :d"), {"d": doc_id})
+        ).scalar_one()
+        assert status == "done"
+        report = (
+            await session.execute(
+                text("SELECT markdown FROM reports WHERE document_id = :d"), {"d": doc_id}
+            )
+        ).scalar_one()
+        assert "Проверено оператором." in report
+
+
+async def test_review_reject_action(
+    client: httpx.AsyncClient, db_session: Any, infra: dict[str, str]
+) -> None:
+    admin_email = f"admin-{uuid.uuid4().hex[:8]}@test.local"
+    await _seed_admin(db_session, email=admin_email, password="adminpass-123")
+    await _login_as(client, admin_email, "adminpass-123")
+
+    user_id, doc_id = await _seed_manual_review_doc(
+        infra, created_at=dt.datetime(2030, 7, 1, tzinfo=dt.UTC)
+    )
+
+    sess = _session_factory()
+    async with sess() as session:
+        from contract_check.core.db.repositories import CreditsRepository
+
+        await CreditsRepository(session).reserve(user_id, document_id=doc_id)
+        await session.commit()
+
+    r = await client.post(
+        f"/admin/review/{doc_id}/reject",
+        data={"reason": "Некорректный формат"},
+        headers={"HX-Request": "true"},
+        follow_redirects=False,
+    )
+    assert r.status_code == 303
+    assert "/admin/review" in r.headers["location"]
+
+    async with sess() as session:
+        status = (
+            await session.execute(text("SELECT status FROM documents WHERE id = :d"), {"d": doc_id})
+        ).scalar_one()
+        assert status == "failed"
+        balance = (
+            await session.execute(
+                text("SELECT credits_left FROM users WHERE id = :u"), {"u": user_id}
+            )
+        ).scalar_one()
+        assert balance == 5
+
+
+async def test_review_send_to_analysis_action(
+    client: httpx.AsyncClient, db_session: Any, infra: dict[str, str]
+) -> None:
+    admin_email = f"admin-{uuid.uuid4().hex[:8]}@test.local"
+    await _seed_admin(db_session, email=admin_email, password="adminpass-123")
+    await _login_as(client, admin_email, "adminpass-123")
+
+    _user_id, doc_id = await _seed_manual_review_doc(
+        infra, created_at=dt.datetime(2030, 8, 1, tzinfo=dt.UTC)
+    )
+
+    # Clean the shared analyze queue so we can consume exactly one message.
+    connection = await aio_pika.connect(infra["rabbitmq_url"])
+    try:
+        channel = await connection.channel()
+        queue = await channel.declare_queue(QUEUE_ANALYZE, passive=True)
+        await queue.purge()
+
+        r = await client.post(
+            f"/admin/review/{doc_id}/send-to-analysis",
+            headers={"HX-Request": "true"},
+            follow_redirects=False,
+        )
+        assert r.status_code == 303
+
+        msg = await queue.get(timeout=5)
+        body = msg.body.decode("utf-8")
+        analyze_msg = AnalyzeRequested.model_validate_json(body)
+        assert analyze_msg.document_id == doc_id
+        assert analyze_msg.attempt == 0
+        assert analyze_msg.prescreen_meta is not None
+        await msg.ack()
+    finally:
+        await connection.close()
+
+    sess = _session_factory()
+    async with sess() as session:
+        status = (
+            await session.execute(text("SELECT status FROM documents WHERE id = :d"), {"d": doc_id})
+        ).scalar_one()
+        assert status == "analyzing"

tests/integration/test_review_actions.py

@@ -0,0 +1,368 @@
+"""Integration tests for manual-review queue actions (issue 018).
+
+These are the "worker handler seam" tests: actions are driven in-process against
+real Postgres / RabbitMQ / MinIO, and the analyze handler is exercised with a stub
+LLM provider to prove that ``send_to_analysis`` produces a real Report.
+"""
+
+from __future__ import annotations
+
+import uuid
+from decimal import Decimal
+from typing import Any
+
+import pytest
+from sqlalchemy import text
+
+from contract_check.core.analysis.report_schema import Finding
+from contract_check.core.db.repositories import (
+    CreditsRepository,
+    DocumentRepository,
+    ReportRepository,
+)
+from contract_check.core.db.session import create_session_factory
+from contract_check.core.llm.port import AnalysisResult
+from contract_check.core.mq.messages import AnalyzeRequested
+from contract_check.core.mq.topology import RK_ANALYZE
+from contract_check.core.review.actions import (
+    ReviewActionError,
+    complete,
+    reject,
+    send_to_analysis,
+)
+from contract_check.core.s3 import extracted_key
+from contract_check.core.s3.minio_storage import MinioStorage
+from contract_check.worker_analyze.handler import AnalyzeHandler
+
+pytestmark = pytest.mark.integration
+
+
+_TEST_TEXT = "Договор поставки. " * 50
+
+
+class FakePublisher:
+    """Captures the message published by ``send_to_analysis``."""
+
+    def __init__(self) -> None:
+        self.messages: list[tuple[Any, str]] = []
+
+    async def publish(self, message: Any, routing_key: str) -> None:
+        self.messages.append((message, routing_key))
+
+
+class StubProvider:
+    """LLMProvider returning one fixed finding."""
+
+    async def analyze(self, text: str, *, checklist: str) -> AnalysisResult:
+        return AnalysisResult(
+            findings=[
+                Finding(
+                    checklist_id="penalties",
+                    severity="high",
+                    quote="штраф 0,5%",
+                    section_ref="п. 6.3",
+                    risk="высокая неустойка",
+                    recommendation="ограничить",
+                )
+            ],
+            model_used="stub-model",
+            prompt_tokens=10,
+            eval_tokens=20,
+            latency_sec=0.1,
+        )
+
+    async def extract_prescreen(self, text: str, *, max_chars: int | None = None) -> dict[str, Any]:
+        raise NotImplementedError
+
+    async def aclose(self) -> None:
+        pass
+
+
+def _storage(infra: dict[str, str]) -> MinioStorage:
+    return MinioStorage.from_endpoint_url(
+        endpoint_url=infra["s3_endpoint_url"],
+        access_key=infra["s3_access_key"],
+        secret_key=infra["s3_secret_key"],
+        bucket="contract-check-docs",
+    )
+
+
+def _session_factory() -> Any:
+    return create_session_factory()
+
+
+async def _seed_user(session: Any, *, telegram_id: int, credits: int = 5) -> uuid.UUID:
+    result = await session.execute(
+        text(
+            "INSERT INTO users (telegram_id, credits_left) VALUES (:t, :c) "
+            "ON CONFLICT (telegram_id) DO UPDATE SET credits_left = :c "
+            "RETURNING id"
+        ),
+        {"t": telegram_id, "c": credits},
+    )
+    return result.scalar_one()
+
+
+async def _seed_manual_review_doc(
+    infra: dict[str, str],
+    *,
+    telegram_id: int | None = None,
+    document_id: uuid.UUID,
+    confidence: float = 0.5,
+    total_amount: Decimal = Decimal("100000.00"),
+) -> tuple[uuid.UUID, str, uuid.UUID]:
+    """Create user + extracted text in MinIO + manual_review document + prescreen result."""
+    if telegram_id is None:
+        telegram_id = int(uuid.uuid4().int % 1_000_000_000)
+    sess = _session_factory()
+    store = _storage(infra)
+    async with sess() as session:
+        user_id = await _seed_user(session, telegram_id=telegram_id)
+        await session.commit()
+
+    ext_key = extracted_key(str(user_id), str(document_id))
+    await store.put(ext_key, _TEST_TEXT.encode("utf-8"), content_type="text/plain; charset=utf-8")
+
+    correlation_id = uuid.uuid4()
+    async with sess() as session:
+        await session.execute(
+            text(
+                "INSERT INTO documents "
+                "(id, user_id, s3_key, extracted_s3_key, filename, mime, bytes, status, stage) "
+                "VALUES (:id, :uid, :s3, :ext, 'contract.pdf', 'application/pdf', :bytes, "
+                " 'manual_review', 'manual_review')"
+            ),
+            {
+                "id": document_id,
+                "uid": user_id,
+                "s3": f"users/{user_id}/docs/{document_id}.pdf",
+                "ext": ext_key,
+                "bytes": len(_TEST_TEXT),
+            },
+        )
+        await session.execute(
+            text(
+                "INSERT INTO prescreen_results "
+                "(id, document_id, correlation_id, contract_type, party_a, party_b, "
+                " total_amount, currency, start_date, end_date, has_penalty_clause, "
+                " has_termination_clause, has_arbitration, confidence_score, routing_decision, "
+                " prescreened_at, auto_summary, auto_findings, extractor_version) "
+                "VALUES (:id, :d, :cid, 'supply', 'ООО Продавец', 'ООО Покупатель', "
+                " :ta, 'RUB', '2025-01-01', '2025-12-31', TRUE, TRUE, FALSE, "
+                " :cs, 'manual_review', now(), 'summary', '[{\"note\":\"ok\"}]', 'heuristic-v2')"
+            ),
+            {
+                "id": uuid.uuid4(),
+                "d": document_id,
+                "cid": correlation_id,
+                "ta": total_amount,
+                "cs": confidence,
+            },
+        )
+        await session.commit()
+    return user_id, ext_key, correlation_id
+
+
+async def _doc_state(session: Any, document_id: uuid.UUID) -> tuple[str, str, bool]:
+    result = await session.execute(
+        text("SELECT status, stage, refunded FROM documents WHERE id = :d"),
+        {"d": document_id},
+    )
+    return result.one()
+
+
+async def test_reject_credits_refunds_once(infra: dict[str, str]) -> None:
+    sess = _session_factory()
+    document_id = uuid.uuid4()
+    user_id, _ext_key, _correlation_id = await _seed_manual_review_doc(
+        infra, document_id=document_id
+    )
+
+    async with sess() as session:
+        await CreditsRepository(session).reserve(user_id, document_id=document_id)
+        await session.commit()
+
+    async with sess() as session:
+        await reject(session, document_id=document_id, reason="Некорректный договор")
+
+    async with sess() as session:
+        status, stage, refunded = await _doc_state(session, document_id)
+        assert status == "failed"
+        assert stage == "review_rejected"
+        assert refunded is True
+
+        balance = await CreditsRepository(session).get_balance(user_id)
+        assert balance == 5
+
+        events = await session.execute(
+            text("SELECT kind, delta FROM credit_events WHERE user_id = :u AND document_id = :d"),
+            {"u": user_id, "d": document_id},
+        )
+        rows = events.all()
+        assert any(kind == "reserve" for kind, _ in rows)
+        assert any(kind == "refund_auto" and delta == 1 for kind, delta in rows)
+
+    # Second reject is idempotent via status, not compensation again.
+    async with sess() as session:
+        with pytest.raises(ReviewActionError):
+            await reject(session, document_id=document_id, reason="")
+
+
+async def test_reject_quota_restores_slot_no_credit_refund(infra: dict[str, str]) -> None:
+    sess = _session_factory()
+    document_id = uuid.uuid4()
+    user_id, _ext_key, _correlation_id = await _seed_manual_review_doc(
+        infra, document_id=document_id
+    )
+
+    subscription_id = uuid.uuid4()
+    async with sess() as session:
+        await session.execute(
+            text(
+                "INSERT INTO subscriptions "
+                "(id, user_id, plan_code, status, current_period_start, current_period_end) "
+                "VALUES (:id, :u, 'lite', 'active', now() - interval '1 day', now() + interval '30 days')"
+            ),
+            {"id": subscription_id, "u": user_id},
+        )
+        await session.execute(
+            text(
+                "INSERT INTO quota_usage (user_id, subscription_id, document_id) "
+                "VALUES (:u, :s, :d)"
+            ),
+            {"u": user_id, "s": subscription_id, "d": document_id},
+        )
+        await session.commit()
+
+    async with sess() as session:
+        await reject(session, document_id=document_id, reason="Неподходящий формат")
+
+    async with sess() as session:
+        status, stage, refunded = await _doc_state(session, document_id)
+        assert status == "failed"
+        assert stage == "review_rejected"
+        assert refunded is True
+
+        quota = await session.execute(
+            text("SELECT id FROM quota_usage WHERE document_id = :d"),
+            {"d": document_id},
+        )
+        assert quota.first() is None
+
+        events = await session.execute(
+            text("SELECT kind, delta FROM credit_events WHERE user_id = :u AND document_id = :d"),
+            {"u": user_id, "d": document_id},
+        )
+        assert not any(kind == "refund_auto" for kind, _ in events.all())
+
+        balance = await CreditsRepository(session).get_balance(user_id)
+        assert balance == 5  # default seeded credits, untouched
+
+
+async def test_complete_writes_report_and_done(infra: dict[str, str]) -> None:
+    sess = _session_factory()
+    document_id = uuid.uuid4()
+    user_id, _ext_key, _correlation_id = await _seed_manual_review_doc(
+        infra, document_id=document_id
+    )
+
+    async with sess() as session:
+        await complete(session, document_id=document_id, note="Договор типовой, рисков нет.")
+
+    async with sess() as session:
+        status, stage, _refunded = await _doc_state(session, document_id)
+        assert status == "done"
+        assert stage == "review_completed"
+
+        report = await ReportRepository(session).get_by_document_id(document_id)
+        assert report is not None
+        assert report.model_used == "admin-complete"
+        assert "Договор типовой, рисков нет." in report.markdown
+        assert report.prescreen_result_id is not None
+        assert report.prescreen_meta is not None
+
+
+async def test_send_to_analysis_publishes_message_and_handler_finishes(
+    infra: dict[str, str],
+) -> None:
+    sess = _session_factory()
+    document_id = uuid.uuid4()
+    user_id, ext_key, _correlation_id = await _seed_manual_review_doc(
+        infra, document_id=document_id
+    )
+
+    fake_publisher = FakePublisher()
+
+    async with sess() as session:
+        await send_to_analysis(session, fake_publisher, document_id=document_id)
+
+    assert len(fake_publisher.messages) == 1
+    msg, routing_key = fake_publisher.messages[0]
+    assert isinstance(msg, AnalyzeRequested)
+    assert msg.document_id == document_id
+    assert msg.user_id == user_id
+    assert msg.extracted_s3_key == ext_key
+    assert msg.attempt == 0
+    assert msg.prescreen_meta is not None
+    assert msg.prescreen_result_id is not None
+    assert routing_key == RK_ANALYZE
+
+    async with sess() as session:
+        status, stage, _refunded = await _doc_state(session, document_id)
+        assert status == "analyzing"
+        assert stage == "queued_analyze"
+
+    # Drive the analyze worker with the same message + stub provider.
+    analyze_handler = AnalyzeHandler(session_factory=sess, provider=StubProvider())
+    try:
+        await analyze_handler.handle(msg)
+    finally:
+        await analyze_handler.aclose()
+
+    async with sess() as session:
+        status, stage, _refunded = await _doc_state(session, document_id)
+        assert status == "done"
+        assert stage == "done"
+
+        report = await ReportRepository(session).get_by_document_id(document_id)
+        assert report is not None
+        assert len(report.content_json.get("findings", [])) == 1
+
+
+async def test_send_to_analysis_rolls_back_status_on_publish_failure(
+    infra: dict[str, str],
+) -> None:
+    sess = _session_factory()
+    document_id = uuid.uuid4()
+    _user_id, _ext_key, _correlation_id = await _seed_manual_review_doc(
+        infra, document_id=document_id
+    )
+
+    class FailingPublisher:
+        async def publish(self, message: Any, routing_key: str) -> None:
+            raise RuntimeError("broker down")
+
+    async with sess() as session:
+        with pytest.raises(RuntimeError):
+            await send_to_analysis(session, FailingPublisher(), document_id=document_id)
+
+    async with sess() as session:
+        status, stage, _refunded = await _doc_state(session, document_id)
+        assert status == "manual_review"
+        assert stage == "manual_review"
+
+
+async def test_action_requires_manual_review_status(infra: dict[str, str]) -> None:
+    sess = _session_factory()
+    document_id = uuid.uuid4()
+    _user_id, _ext_key, _correlation_id = await _seed_manual_review_doc(
+        infra, document_id=document_id
+    )
+
+    async with sess() as session:
+        await DocumentRepository(session).update_status(document_id, status="done")
+        await session.commit()
+
+    async with sess() as session:
+        with pytest.raises(ReviewActionError):
+            await complete(session, document_id=document_id, note="note")

tests/unit/test_review_repository.py

@@ -0,0 +1,276 @@
+"""Repository tests for the manual-review queue query.
+
+Backs issue 017: queue rows are documents in ``manual_review`` joined with the
+latest prescreen result, filterable and paginated. These tests hit the Docker
+Compose Postgres stack and roll back after each test.
+"""
+
+from __future__ import annotations
+
+import datetime as dt
+import uuid
+from decimal import Decimal
+
+import pytest
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from contract_check.core.db.repositories import ReviewQueueRepository
+
+pytestmark = [pytest.mark.unit, pytest.mark.usefixtures("db_session")]
+
+
+@pytest.fixture(autouse=True)
+async def _clean_queue_before_each(db_session: AsyncSession) -> None:
+    """Give every queue test a clean manual-review slice inside the savepoint."""
+    await _clean_manual_review(db_session)
+
+
+async def _clean_manual_review(session: AsyncSession) -> None:
+    """Remove leftover manual-review rows from earlier integration runs.
+
+    Runs inside the per-test nested transaction; the enclosing savepoint rolls
+    back the cleanup at the end of the test, so the shared dev database keeps
+    whatever was there before.
+    """
+    await session.execute(
+        text(
+            "DELETE FROM credit_events WHERE document_id IN "
+            "(SELECT id FROM documents WHERE status = 'manual_review')"
+        )
+    )
+    await session.execute(text("DELETE FROM documents WHERE status = 'manual_review'"))
+
+
+async def _seed_user(session: AsyncSession, *, telegram_id: int = 0) -> uuid.UUID:
+    if telegram_id == 0:
+        telegram_id = int.from_bytes(uuid.uuid4().bytes[:4], "big", signed=True)
+    result = await session.execute(
+        text("INSERT INTO users (telegram_id, credits_left) VALUES (:t, 5) RETURNING id"),
+        {"t": telegram_id},
+    )
+    return result.scalar_one()
+
+
+_TEST_EPOCH = dt.datetime(2030, 1, 1, tzinfo=dt.UTC)
+
+
+async def _seed_manual_review_doc(
+    session: AsyncSession,
+    *,
+    user_id: uuid.UUID,
+    created_at: dt.datetime | None = None,
+    confidence: Decimal | float = Decimal("0.5"),
+    total_amount: Decimal | float | None = Decimal("100000.00"),
+) -> tuple[uuid.UUID, uuid.UUID]:
+    document_id = uuid.uuid4()
+    correlation_id = uuid.uuid4()
+    if created_at is None:
+        created_at = _TEST_EPOCH
+    await session.execute(
+        text(
+            "INSERT INTO documents "
+            "(id, user_id, s3_key, extracted_s3_key, filename, mime, bytes, status, stage) "
+            "VALUES (:id, :uid, :s3, :ext, 'contract.pdf', 'application/pdf', 1000, "
+            " 'manual_review', 'manual_review')"
+        ),
+        {
+            "id": document_id,
+            "uid": user_id,
+            "s3": f"users/{user_id}/docs/{document_id}.pdf",
+            "ext": f"users/{user_id}/docs/{document_id}.txt",
+        },
+    )
+    await session.execute(
+        text(
+            "INSERT INTO prescreen_results "
+            "(id, document_id, correlation_id, contract_type, party_a, party_b, "
+            " total_amount, currency, start_date, end_date, has_penalty_clause, "
+            " has_termination_clause, has_arbitration, confidence_score, routing_decision, "
+            " prescreened_at, auto_summary, auto_findings, extractor_version) "
+            "VALUES (:id, :d, :cid, 'supply', 'ООО Продавец', 'ООО Покупатель', "
+            " :ta, 'RUB', '2025-01-01', '2025-12-31', TRUE, TRUE, FALSE, "
+            " :cs, 'manual_review', now(), 'summary', '[{\"note\":\"ok\"}]', 'heuristic-v2')"
+        ),
+        {
+            "id": uuid.uuid4(),
+            "d": document_id,
+            "cid": correlation_id,
+            "ta": total_amount,
+            "cs": confidence,
+        },
+    )
+    # created_at is server-default now(); override it for ordering tests.
+    if created_at:
+        await session.execute(
+            text("UPDATE documents SET created_at = :ca WHERE id = :id"),
+            {"ca": created_at, "id": document_id},
+        )
+    await session.commit()
+    return document_id, correlation_id
+
+
+async def test_list_returns_manual_review_documents(db_session: AsyncSession) -> None:
+    repo = ReviewQueueRepository(db_session)
+    user_id = await _seed_user(db_session)
+    doc_id, _ = await _seed_manual_review_doc(db_session, user_id=user_id)
+
+    rows, total = await repo.list_queue(
+        date_from=dt.date(2030, 1, 1),
+        date_to=dt.date(2030, 12, 31),
+        limit=10,
+        offset=0,
+    )
+    assert total == 1
+    assert len(rows) == 1
+    assert rows[0].document_id == doc_id
+    assert rows[0].party_a == "ООО Продавец"
+    assert rows[0].confidence_score == Decimal("0.5")
+
+
+async def test_list_orders_newest_first(db_session: AsyncSession) -> None:
+    repo = ReviewQueueRepository(db_session)
+    user_id = await _seed_user(db_session)
+    older = await _seed_manual_review_doc(
+        db_session, user_id=user_id, created_at=dt.datetime(2030, 1, 1, tzinfo=dt.UTC)
+    )
+    newer = await _seed_manual_review_doc(
+        db_session, user_id=user_id, created_at=dt.datetime(2030, 6, 1, tzinfo=dt.UTC)
+    )
+
+    rows, _ = await repo.list_queue(
+        date_from=dt.date(2030, 1, 1),
+        date_to=dt.date(2030, 12, 31),
+        limit=10,
+        offset=0,
+    )
+    assert [r.document_id for r in rows] == [newer[0], older[0]]
+
+
+async def test_list_ignores_non_manual_review(db_session: AsyncSession) -> None:
+    repo = ReviewQueueRepository(db_session)
+    user_id = await _seed_user(db_session)
+    manual = await _seed_manual_review_doc(db_session, user_id=user_id)
+
+    ignored_id = uuid.uuid4()
+    await db_session.execute(
+        text(
+            "INSERT INTO documents "
+            "(id, user_id, s3_key, filename, mime, bytes, status) "
+            "VALUES (:id, :uid, :s3, 'x.pdf', 'application/pdf', 1, 'analyzing')"
+        ),
+        {"id": ignored_id, "uid": user_id, "s3": f"users/{user_id}/docs/{ignored_id}.pdf"},
+    )
+    await db_session.execute(
+        text(
+            "INSERT INTO prescreen_results "
+            "(id, document_id, correlation_id, routing_decision, prescreened_at) "
+            "VALUES (:id, :d, :cid, 'manual_review', now())"
+        ),
+        {"id": uuid.uuid4(), "d": ignored_id, "cid": uuid.uuid4()},
+    )
+    await db_session.commit()
+
+    rows, total = await repo.list_queue(
+        date_from=dt.date(2030, 1, 1),
+        date_to=dt.date(2030, 12, 31),
+        limit=10,
+        offset=0,
+    )
+    assert total == 1
+    assert rows[0].document_id == manual[0]
+
+
+async def test_list_pagination(db_session: AsyncSession) -> None:
+    repo = ReviewQueueRepository(db_session)
+    user_id = await _seed_user(db_session)
+    for i in range(3):
+        await _seed_manual_review_doc(
+            db_session,
+            user_id=user_id,
+            created_at=dt.datetime(2030, 1, 1 + i, tzinfo=dt.UTC),
+        )
+
+    rows, total = await repo.list_queue(
+        date_from=dt.date(2030, 1, 1),
+        date_to=dt.date(2030, 12, 31),
+        limit=2,
+        offset=0,
+    )
+    assert total == 3
+    assert len(rows) == 2
+
+    rows2, _ = await repo.list_queue(
+        date_from=dt.date(2030, 1, 1),
+        date_to=dt.date(2030, 12, 31),
+        limit=2,
+        offset=2,
+    )
+    assert len(rows2) == 1
+
+
+async def test_list_filters_by_date(db_session: AsyncSession) -> None:
+    repo = ReviewQueueRepository(db_session)
+    user_id = await _seed_user(db_session)
+    inside = await _seed_manual_review_doc(
+        db_session, user_id=user_id, created_at=dt.datetime(2030, 3, 15, tzinfo=dt.UTC)
+    )
+    await _seed_manual_review_doc(
+        db_session, user_id=user_id, created_at=dt.datetime(2030, 6, 1, tzinfo=dt.UTC)
+    )
+
+    rows, total = await repo.list_queue(
+        date_from=dt.date(2030, 3, 1), date_to=dt.date(2030, 3, 31), limit=10, offset=0
+    )
+    assert total == 1
+    assert rows[0].document_id == inside[0]
+
+
+async def test_list_filters_by_confidence(db_session: AsyncSession) -> None:
+    repo = ReviewQueueRepository(db_session)
+    user_id = await _seed_user(db_session)
+    low = await _seed_manual_review_doc(db_session, user_id=user_id, confidence=Decimal("0.3"))
+    high = await _seed_manual_review_doc(db_session, user_id=user_id, confidence=Decimal("0.9"))
+
+    rows, _ = await repo.list_queue(
+        date_from=dt.date(2030, 1, 1),
+        date_to=dt.date(2030, 12, 31),
+        conf_min=Decimal("0.5"),
+        conf_max=Decimal("1.0"),
+        limit=10,
+        offset=0,
+    )
+    assert len(rows) == 1
+    assert rows[0].document_id == high[0]
+
+    rows, _ = await repo.list_queue(
+        date_from=dt.date(2030, 1, 1),
+        date_to=dt.date(2030, 12, 31),
+        conf_min=Decimal("0.1"),
+        conf_max=Decimal("0.5"),
+        limit=10,
+        offset=0,
+    )
+    assert len(rows) == 1
+    assert rows[0].document_id == low[0]
+
+
+async def test_get_item_returns_detail_fields(db_session: AsyncSession) -> None:
+    repo = ReviewQueueRepository(db_session)
+    user_id = await _seed_user(db_session)
+    doc_id, correlation_id = await _seed_manual_review_doc(
+        db_session, user_id=user_id, total_amount=Decimal("250000.00")
+    )
+
+    item = await repo.get_item(doc_id)
+    assert item is not None
+    assert item.document_id == doc_id
+    assert item.correlation_id == correlation_id
+    assert item.user_id == user_id
+    assert item.total_amount == Decimal("250000.00")
+    assert item.extracted_s3_key == f"users/{user_id}/docs/{doc_id}.txt"
+
+
+async def test_get_item_returns_none_for_missing(db_session: AsyncSession) -> None:
+    repo = ReviewQueueRepository(db_session)
+    assert await repo.get_item(uuid.uuid4()) is None