feat(contract): add opt-in MAX configuration and lifecycle

Refs #81
2026-10-04 15:51:35 +03:00
15 changed files with 10 additions and 879 deletions
@@ -1,115 +0,0 @@
 """Server API for a future MAX connect UI; webhook processing is supplied by the channel adapter."""
 import hmac
 import json
 import re
 from datetime import datetime
 from typing import Annotated, Literal
 from fastapi import APIRouter, Depends, HTTPException, Request
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 from pydantic import BaseModel, ConfigDict
 from sqlalchemy import select
 from sqlalchemy.exc import SQLAlchemyError
 from app.config import get_settings
 from app.core import clock
 from app.db import repo
 from app.db.max_notifications import MaxBinding, MaxDelivery
 from app.db.session import get_clinical_session, get_identity_session
 from app.integrations import max_store
 router = APIRouter(tags=["max-notifications"])
 bearer = HTTPBearer(auto_error=False)
 async def patient_token(auth: Annotated[HTTPAuthorizationCredentials | None, Depends(bearer)]) -> str:
    if auth is None or auth.scheme.lower() != "bearer" or not re.fullmatch(r"[A-Za-z0-9_-]{1,64}", auth.credentials):
        raise HTTPException(401, "patient authentication required", headers={"WWW-Authenticate": "Bearer"})
    async with get_clinical_session() as session:
        if await repo.get_profile(session, auth.credentials) is None:
            raise HTTPException(401, "patient authentication required")
    return auth.credentials
 PatientToken = Annotated[str, Depends(patient_token)]
 class MaxStatus(BaseModel):
    available: bool
    connected: bool
    last_delivery: str | None = None
 class LinkConsent(BaseModel):
    model_config = ConfigDict(extra="forbid")
    consent: Literal[True]
 class MaxLink(BaseModel):
    url: str
    expires_at: datetime
@router.get("/me/notifications/max", response_model=MaxStatus)
 async def status(token: PatientToken) -> MaxStatus:
    async with get_identity_session() as session:
        binding = await session.get(MaxBinding, token)
        last = await session.scalar(
            select(MaxDelivery.state)
            .where(MaxDelivery.token == token)
            .order_by(MaxDelivery.created_at.desc(), MaxDelivery.id.desc())
            .limit(1)
        )
    return MaxStatus(
        available=get_settings().max_enabled, connected=bool(binding and binding.enabled), last_delivery=last
    )
@router.post("/me/notifications/max/link", response_model=MaxLink)
 async def link(body: LinkConsent, token: PatientToken) -> MaxLink:
    settings = get_settings()
    if not settings.max_enabled:
        raise HTTPException(503, "MAX is not configured")
    if not await max_store.active_patient(token):
        raise HTTPException(403, "active consent and adult confirmation required")
    async with get_identity_session() as session:
        raw, expires = await max_store.issue_code(session, token, clock.now(), settings.max_link_ttl_seconds)
    return MaxLink(url=f"https://max.ru/{settings.max_bot_username}?start={raw}", expires_at=expires)
@router.delete("/me/notifications/max", response_model=MaxStatus)
 async def unlink(token: PatientToken) -> MaxStatus:
    async with get_identity_session() as session:
        await max_store.disconnect(session, token)
    return MaxStatus(available=get_settings().max_enabled, connected=False)
@router.post("/webhooks/max")
 async def webhook(request: Request) -> dict[str, bool]:
    settings = get_settings()
    if not settings.max_enabled:
        raise HTTPException(503, "MAX is not configured")
    expected = settings.max_webhook_secret.get_secret_value().encode()
    supplied = request.headers.get("X-Max-Bot-Api-Secret", "").encode()
    if not expected or not hmac.compare_digest(supplied, expected):
        raise HTTPException(403, "invalid webhook secret")
    raw = bytearray()
    async for chunk in request.stream():
        raw.extend(chunk)
        if len(raw) > 65536:
            raise HTTPException(413, "event is too large")
    try:
        event = json.loads(raw)
    except (ValueError, UnicodeError) as exc:
        raise HTTPException(400, "invalid JSON") from exc
    if not isinstance(event, dict):
        raise HTTPException(422, "event must be an object")
    from app.channels.max_bot import handle_update
    try:
        await handle_update(event)
    except SQLAlchemyError:
        # Let MAX retry; do not log DB exception parameters containing identity data.
        raise HTTPException(503, "notification storage unavailable") from None
    return {"success": True}
@@ -15,7 +15,7 @@ from sqlalchemy import select
 from app.api import demo as console
 from app.api import doctor_views as views
-from app.api import fixtures, max_notifications
+from app.api import fixtures
 from app.api.patient_web import router as patient_web_router
 from app.api.questionnaire import router as questionnaire_router
 from app.api.schemas import (
@@ -46,7 +46,6 @@ from app.scheduler import jobs as scheduler
 from app.services.monitoring import refresh_status
 router = APIRouter(prefix="/api/v1")
 router.include_router(max_notifications.router)
 router.include_router(questionnaire_router)
 router.include_router(patient_web_router)
@@ -37,7 +37,6 @@ class Settings(BaseSettings):
    openrouter_api_key: str = ""
    openrouter_model: str = "google/gemini-2.5-flash-lite"
    openrouter_base_url: str = "https://openrouter.ai/api/v1"
    gigachat_agent_model: str = "GigaChat-2-Pro"  # function calling quality matters more than price here
    chat_agent_timeout_s: float = 25.0
    chat_agent_max_steps: int = 5
@@ -5,7 +5,6 @@ The only module (besides registration code) that touches both schemas — and on
 from app.db import audit as _audit  # noqa: F401  (registers audit_log in clinical metadata)
 from app.db import intake as _intake  # noqa: F401
 from app.db import max_notifications as _max_notifications  # noqa: F401
 from app.db import scheduler as _scheduler  # noqa: F401
 from app.db.clinical import ClinicalBase
 from app.db.identity import IdentityBase
@@ -1,50 +0,0 @@
 """MAX identifiers and delivery metadata live only in the identity DB; no medical message bodies."""
 from datetime import datetime
 from sqlalchemy import BigInteger, Boolean, Integer, String
 from sqlalchemy.orm import Mapped, mapped_column
 from app.db.identity import IdentityBase
 from app.db.types import TZDateTime
 class MaxBinding(IdentityBase):
    __tablename__ = "max_bindings"
    # No FK: anonymous web profiles intentionally have no patients row in identity.
    token: Mapped[str] = mapped_column(String(64), primary_key=True)
    user_id: Mapped[int] = mapped_column(BigInteger, unique=True)
    enabled: Mapped[bool] = mapped_column(Boolean, default=True)
    connected_at: Mapped[datetime] = mapped_column(TZDateTime)
 class MaxLinkCode(IdentityBase):
    __tablename__ = "max_link_codes"
    digest: Mapped[str] = mapped_column(String(64), primary_key=True)
    token: Mapped[str] = mapped_column(String(64), unique=True)
    expires_at: Mapped[datetime] = mapped_column(TZDateTime, index=True)
    used_at: Mapped[datetime | None] = mapped_column(TZDateTime)
 class MaxReceipt(IdentityBase):
    __tablename__ = "max_webhook_receipts"
    digest: Mapped[str] = mapped_column(String(64), primary_key=True)
    received_at: Mapped[datetime] = mapped_column(TZDateTime, index=True)
 class MaxDelivery(IdentityBase):
    __tablename__ = "max_deliveries"
    id: Mapped[str] = mapped_column(String(64), primary_key=True)
    token: Mapped[str | None] = mapped_column(String(64), index=True)
    user_id: Mapped[int] = mapped_column(BigInteger)
    kind: Mapped[str] = mapped_column(String(24))
    state: Mapped[str] = mapped_column(String(16), default="queued", index=True)
    attempts: Mapped[int] = mapped_column(Integer, default=0)
    created_at: Mapped[datetime] = mapped_column(TZDateTime)
    next_attempt_at: Mapped[datetime] = mapped_column(TZDateTime, index=True)
    lease: Mapped[str | None] = mapped_column(String(32))
    error_code: Mapped[str | None] = mapped_column(String(32))
@@ -1,201 +0,0 @@
 """Registration boundary and durable MAX queue. Callers own the identity transaction.
 Only this boundary joins a pseudonymous profile to a MAX recipient. Clinical modules never import it.
 Queue rows contain a template name, never replies, scores, prescriptions or patient-auth tokens in URLs.
 """
 import hashlib
 import secrets
 from datetime import datetime, timedelta
 from sqlalchemy import delete, select, update
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 from app.db import repo
 from app.db.clinical import ScaleResult
 from app.db.intake import QuestionnaireRecord
 from app.db.max_notifications import MaxBinding, MaxDelivery, MaxLinkCode
 from app.db.session import get_clinical_session
 MAX_ATTEMPTS = 6
 LEASE_SECONDS = 60
 def digest(value: str) -> str:
    return hashlib.sha256(value.encode()).hexdigest()
 async def active_patient(token: str) -> bool:
    """Same consent/adult gate used by the scheduler, checked again at delivery time."""
    async with get_clinical_session() as session:
        state = await repo.get_dialog_state(session, token)
        return bool(
            await repo.get_profile(session, token)
            and state
            and state.data.get("consent") is True
            and state.data.get("adult") is True
        )
 async def issue_code(session: AsyncSession, token: str, now: datetime, ttl: int) -> tuple[str, datetime]:
    raw = secrets.token_urlsafe(32)
    expires = now + timedelta(seconds=ttl)
    await session.execute(delete(MaxLinkCode).where(MaxLinkCode.token == token))
    session.add(MaxLinkCode(digest=digest(raw), token=token, expires_at=expires))
    await session.flush()
    return raw, expires
 async def code_token(session: AsyncSession, raw: str, now: datetime) -> str | None:
    row = await session.get(MaxLinkCode, digest(raw))
    return row.token if row and row.used_at is None and row.expires_at > now else None
 async def bind(session: AsyncSession, raw: str, user_id: int, now: datetime) -> str | None:
    """Atomically consume a code. Existing active identities cannot be silently reassigned."""
    token = await code_token(session, raw, now)
    if token is None:
        return None
    owner = await session.scalar(select(MaxBinding).where(MaxBinding.user_id == user_id))
    binding = await session.get(MaxBinding, token)
    if owner and owner.token != token and owner.enabled:
        return None
    if binding and binding.user_id != user_id and binding.enabled:
        return None
    used = await session.execute(
        update(MaxLinkCode)
        .where(MaxLinkCode.digest == digest(raw), MaxLinkCode.used_at.is_(None), MaxLinkCode.expires_at > now)
        .values(used_at=now)
        .returning(MaxLinkCode.token)
    )
    if used.scalar_one_or_none() is None:
        return None
    if owner and owner.token != token:
        await session.delete(owner)
        await session.flush()
    if binding is None:
        binding = MaxBinding(token=token, user_id=user_id, connected_at=now)
        session.add(binding)
    # Cancel stale deliveries even when the same account reconnects later.
    await cancel_pending(session, token)
    binding.user_id, binding.enabled, binding.connected_at = user_id, True, now
    await session.flush()
    return token
 async def cancel_pending(session: AsyncSession, token: str) -> None:
    await session.execute(
        update(MaxDelivery)
        .where(MaxDelivery.token == token, MaxDelivery.state.in_(("queued", "sending")))
        .values(state="cancelled", lease=None)
    )
 async def disconnect(session: AsyncSession, token: str) -> None:
    binding = await session.get(MaxBinding, token)
    if binding:
        binding.enabled = False
    await session.execute(delete(MaxLinkCode).where(MaxLinkCode.token == token))
    await cancel_pending(session, token)
 async def enqueue(
    session: AsyncSession, *, key: str, user_id: int, kind: str, now: datetime, token: str | None = None
 ) -> None:
    if await session.get(MaxDelivery, key) is not None:
        return
    try:
        async with session.begin_nested():
            session.add(
                MaxDelivery(id=key, user_id=user_id, token=token, kind=kind, created_at=now, next_attempt_at=now)
            )
            await session.flush()
    except IntegrityError:
        # Another worker may enqueue the same event between SELECT and INSERT.
        if await session.get(MaxDelivery, key) is None:
            raise
 async def collect_completed(session: AsyncSession, now: datetime) -> None:
    """Recover completion events from committed clinical records, including after a restart.
    No clinical hooks depend on identity. Read only IDs/dates, then enqueue a generic template.
    A binding's activation time bounds history; repeated scans are idempotent.
    """
    bindings = (await session.scalars(select(MaxBinding).where(MaxBinding.enabled.is_(True)))).all()
    for binding in bindings:
        if not await active_patient(binding.token):
            continue
        async with get_clinical_session() as clinical:
            forms = (
                await clinical.scalars(
                    select(QuestionnaireRecord.id).where(
                        QuestionnaireRecord.token == binding.token,
                        QuestionnaireRecord.status == "complete",
                        QuestionnaireRecord.completed_at >= binding.connected_at,
                    )
                )
            ).all()
            scales = (
                await clinical.scalars(
                    select(ScaleResult.id).where(
                        ScaleResult.token == binding.token,
                        ScaleResult.complete.is_(True),
                        ScaleResult.ts >= binding.connected_at,
                    )
                )
            ).all()
        for source, ids in (("form", forms), ("scale", scales)):
            for record_id in ids:
                key = digest(f"completed:{binding.token}:{binding.connected_at.isoformat()}:{source}:{record_id}")
                await enqueue(session, key=key, token=binding.token, user_id=binding.user_id, kind="completed", now=now)
 async def claim(session: AsyncSession, now: datetime) -> MaxDelivery | None:
    """One lease per row; recovery after a worker crash is bounded by the lease deadline."""
    ready = (MaxDelivery.state.in_(("queued", "sending"))) & (MaxDelivery.next_attempt_at <= now)
    row = await session.scalar(
        select(MaxDelivery)
        .where(ready)
        .order_by(MaxDelivery.created_at, MaxDelivery.id)
        .with_for_update(skip_locked=True)
        .limit(1)
    )
    if row is None:
        return None
    if row.attempts >= MAX_ATTEMPTS:
        row.state, row.error_code, row.lease = "failed", "attempt_limit", None
        return None
    lease = secrets.token_hex(16)
    result = await session.execute(
        update(MaxDelivery)
        .where(MaxDelivery.id == row.id, ready)
        .values(
            state="sending",
            attempts=MaxDelivery.attempts + 1,
            lease=lease,
            next_attempt_at=now + timedelta(seconds=LEASE_SECONDS),
        )
        .returning(MaxDelivery.id)
    )
    if result.scalar_one_or_none() is None:
        return None
    await session.refresh(row)
    return row
 async def finish(
    session: AsyncSession,
    row: MaxDelivery,
    *,
    state: str,
    now: datetime,
    error: str | None = None,
    retry_seconds: float = 0,
 ) -> None:
    await session.execute(
        update(MaxDelivery)
        .where(MaxDelivery.id == row.id, MaxDelivery.lease == row.lease, MaxDelivery.state == "sending")
        .values(state=state, lease=None, error_code=error, next_attempt_at=now + timedelta(seconds=retry_seconds))
    )
@@ -22,7 +22,6 @@ from app.config import Settings, get_settings
 from app.core.types import ScrubbedText
 from app.llm.agent.prompts import SYSTEM_PROMPT
 from app.llm.agent.providers import (
    GigaChatToolProvider,
    MockAgentProvider,
    OpenRouterProvider,
    ProviderError,
@@ -50,7 +49,7 @@ def build_provider(settings: Settings | None = None) -> ToolChatProvider | None:
    s = settings or get_settings()
    mode = s.chat_agent
    if mode == "auto":
-        mode = "gigachat" if s.gigachat_credentials else "openrouter" if s.openrouter_api_key else "off"
+        mode = "openrouter" if s.openrouter_api_key else "off"
    if mode == "openrouter":
        if not s.openrouter_api_key:
            log.warning("chat agent: openrouter selected without OPENROUTER_API_KEY — disabled")
@@ -59,16 +58,15 @@ def build_provider(settings: Settings | None = None) -> ToolChatProvider | None:
    if mode == "mock":
        return MockAgentProvider()
    if mode == "gigachat":
-        if not s.gigachat_credentials:
+        # TODO(GigaChat): function calling via the `gigachat` package (functions=..., message.function_call).
-            log.warning("chat agent: gigachat selected without GIGACHAT_CREDENTIALS — disabled")
+        log.warning("chat agent: gigachat provider not implemented yet — disabled")
        return None
        return GigaChatToolProvider(s)
    return None
 async def _chat_with_retry(provider: ToolChatProvider, messages: list[dict[str, Any]]):
    """One retry: OpenRouter occasionally fails a single call (seen live as intermittent ProviderError)."""
-    wire = _wire(messages) if provider.name == "openrouter" else messages  # gigachat needs the tool `name`
+    wire = _wire(messages) if provider.name != "mock" else messages
    try:
        return await provider.chat(wire, TOOL_SPECS)
    except ProviderError as exc:
@@ -141,9 +139,6 @@ async def answer_chat(
    except (ProviderError, TimeoutError) as exc:
        log.error("chat agent provider=%s failed: %s %s", provider.name, type(exc).__name__, exc)
        return None
    except Exception:  # noqa: BLE001 — the agent is optional: any bug must degrade to the extractive answer, not a 500
        log.exception("chat agent provider=%s crashed", provider.name)
        return None
 def _finish(text: str, ctx: ToolContext, provider: str) -> AgentAnswer | None:
@@ -68,10 +68,9 @@ class OpenRouterProvider:
        if resp.status_code != 200:
            # Provider error bodies are generic (no patient text) — log the message to make failures diagnosable.
            try:
-                err = resp.json().get("error", "")
+                detail = str(resp.json().get("error", {}).get("message", ""))[:200]
-                detail = str(err.get("message", "") if isinstance(err, dict) else err)[:200]
+            except ValueError:
-            except (ValueError, AttributeError):
+                detail = ""
                detail = resp.text[:200]
            log.error("openrouter http %s: %s", resp.status_code, detail)
            raise ProviderError(f"http {resp.status_code}")
        try:
@@ -91,88 +90,6 @@ class OpenRouterProvider:
        return AssistantTurn(content=msg.get("content"), tool_calls=calls)
 class GigaChatToolProvider:
    """GigaChat function calling (RF provider, 152-FZ compliant). Converts the OpenAI-style history the agent keeps:
    assistant `tool_calls` → `function_call`, `tool` results → role `function` with the function name.
    GigaChat returns at most one function call per turn — the agent loop handles that naturally."""
    name = "gigachat"
    def __init__(self, settings: Settings, client: Any | None = None) -> None:
        self._settings = settings
        self._client = client
    def _get_client(self) -> Any:
        if self._client is None:
            from gigachat import GigaChat
            s = self._settings
            self._client = GigaChat(
                credentials=s.gigachat_credentials,
                scope=s.gigachat_scope,
                model=s.gigachat_agent_model,
                verify_ssl_certs=s.gigachat_verify_ssl,
                timeout=s.chat_agent_timeout_s,
            )
        return self._client
    @staticmethod
    def to_messages(messages: list[dict[str, Any]]) -> list[Any]:
        from gigachat.models import FunctionCall, Messages, MessagesRole
        out = []
        for m in messages:
            role = m["role"]
            if role == "system":
                out.append(Messages(role=MessagesRole.SYSTEM, content=m["content"]))
            elif role == "user":
                out.append(Messages(role=MessagesRole.USER, content=m["content"]))
            elif role == "assistant":
                calls = m.get("tool_calls") or []
                for c in calls or [None]:
                    fc = None
                    if c is not None:
                        fc = FunctionCall(name=c["function"]["name"], arguments=json.loads(c["function"]["arguments"]))
                    out.append(Messages(role=MessagesRole.ASSISTANT, content=m.get("content") or "", function_call=fc))
            elif role == "tool":
                out.append(Messages(role=MessagesRole.FUNCTION, name=m.get("name"), content=m["content"]))
        return out
    @staticmethod
    def to_functions(tools: list[dict[str, Any]]) -> list[Any]:
        from gigachat.models import Function
        return [Function.model_validate(t["function"]) for t in tools]
    async def chat(self, messages: list[dict[str, Any]], tools: list[dict[str, Any]]) -> AssistantTurn:
        from gigachat.models import Chat
        chat = Chat(
            messages=self.to_messages(messages),
            functions=self.to_functions(tools),
            function_call="auto",
            temperature=0.1,
            max_tokens=900,
        )
        try:
            resp = await self._get_client().achat(chat)
        except Exception as exc:  # noqa: BLE001 — network/auth/HTTP errors of the SDK
            raise ProviderError(type(exc).__name__) from exc
        msg = resp.choices[0].message
        calls = []
        if msg.function_call is not None:
            args = msg.function_call.arguments or {}
            if isinstance(args, str):
                try:
                    args = json.loads(args)
                except json.JSONDecodeError:
                    args = {}
            calls.append(ToolCall(id=msg.function_call.name, name=msg.function_call.name, arguments=args))
        usage = getattr(resp, "usage", None)
        log.info("gigachat agent ok calls=%d tokens=%s", len(calls), getattr(usage, "total_tokens", None))
        return AssistantTurn(content=msg.content or None, tool_calls=calls)
 # ------------------------------------------------------------------------------------------------ offline mock
 _DRUG_WORDS = re.compile(
@@ -1,24 +0,0 @@
 #!/usr/bin/env bash
 # Deploy origin/dev to the shared server (ssh alias `vobot`, dir /srv/ptsr). Run from a clean checkout:
 #   infra/deploy/deploy.sh [git-ref]        default: origin/dev
 # Keeps /srv/ptsr/.env (secrets live only on the server) and the Postgres volume.
 set -euo pipefail
 REF=${1:-origin/dev}
 HOST=${DEPLOY_HOST:-vobot}
 DIR=/srv/ptsr
 git fetch -q origin || true
 SHA=$(git rev-parse --short "$REF")
 echo "deploy $REF ($SHA) -> $HOST:$DIR"
 git archive --format=tar "$REF" | ssh "$HOST" "rm -rf $DIR/app.new && mkdir -p $DIR/app.new && tar -x -C $DIR/app.new"
 ssh "$HOST" bash -s <<SH
 set -euo pipefail
 cd $DIR
 cp .env app.new/.env
 rm -rf app.old; [ -d app ] && mv app app.old; mv app.new app
 cd app
 echo "$SHA" > REVISION
 docker run --rm -v "\$PWD/web:/w" -w /w node:20-alpine sh -c "npm ci --no-audit --no-fund >/dev/null && npm run build >/dev/null"
 docker compose -p ptsr -f docker-compose.yml -f infra/deploy/docker-compose.server.yml up -d --build </dev/null
 for i in \$(seq 1 40); do curl -sf 127.0.0.1:8010/health >/dev/null && break; sleep 3; done
 curl -sf 127.0.0.1:8010/health && echo " api ok"
 SH
@@ -1,30 +0,0 @@
 # Server override (shared host with vobot): loopback-only ports behind host nginx, restart policy, memory caps.
 #   docker compose -p ptsr -f docker-compose.yml -f infra/deploy/docker-compose.server.yml up -d --build
 services:
  postgres:
    ports: !reset []
    restart: unless-stopped
    mem_limit: 512m
  app:
    ports: !override
      - "127.0.0.1:8010:8000"
    restart: unless-stopped
    mem_limit: 768m
  dashboard:
    command:
      - streamlit
      - run
      - dashboard/app.py
      - --server.port=8501
      - --server.address=0.0.0.0
      - --server.baseUrlPath=doctor
      - --browser.gatherUsageStats=false
    environment:
      API_BASE_URL: http://app:8000
      DASHBOARD_PUBLIC_URL: ${PUBLIC_URL:-http://localhost}/doctor
    ports: !override
      - "127.0.0.1:8511:8501"
    restart: unless-stopped
    mem_limit: 768m
@@ -1,52 +0,0 @@
 # meditron.tenvex.ru — PTSRaiAgent prototype (synthetic data only).
 #   /          patient web app (static, web/dist)
 #   /api/      FastAPI (127.0.0.1:8010)
 #   /doctor/   Streamlit doctor dashboard + demo console (127.0.0.1:8511, baseUrlPath=doctor)
 # TLS: after the DNS A-record exists — certbot --nginx -d meditron.tenvex.ru
 map $http_upgrade $ptsr_connection_upgrade {
    default upgrade;
    ''      close;
 }
 server {
    listen 80;
    listen [::]:80;
    server_name meditron.tenvex.ru;
    access_log /var/log/nginx/ptsr-access.log;
    client_max_body_size 5m;
    root /srv/ptsr/app/web/dist;
    index index.html;
    location / {
        try_files $uri $uri/ /index.html;
    }
    location /api/ {
        proxy_pass http://127.0.0.1:8010;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
        proxy_read_timeout 120s;   # chat agent + scheduler replay on clock advance
    }
    location ~ ^/(docs|openapi.json|health)$ {
        proxy_pass http://127.0.0.1:8010;
        proxy_set_header Host $host;
    }
    location /doctor/ {
        proxy_pass http://127.0.0.1:8511;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection $ptsr_connection_upgrade;
        proxy_set_header Host $host;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
        proxy_read_timeout 86400s;
        proxy_buffering off;
    }
 }
@@ -1,103 +0,0 @@
 """MAX endpoints authenticate before reading bindings or accepting events."""
 import sys
 from types import SimpleNamespace
 from unittest.mock import AsyncMock
 from urllib.parse import parse_qs, urlsplit
 import pytest
 from httpx import ASGITransport, AsyncClient
 from sqlalchemy.exc import OperationalError
 from app.config import get_settings
 from app.core import clock
 from app.db import repo, session
 from app.db.bootstrap import init_db
 from app.integrations import max_store
 from app.main import app
 BASE = "/api/v1/me/notifications/max"
 HOOK = "/api/v1/webhooks/max"
 TOKEN = "synthetic-patient"
 AUTH = {"Authorization": f"Bearer {TOKEN}"}
 SECRET = "test-secret-" * 4
@pytest.fixture
 async def client(monkeypatch):
    for key, value in {
        "MAX_ENABLED": "true",
        "MAX_BOT_TOKEN": "fake-token",
        "MAX_BOT_USERNAME": "test_bot",
        "MAX_WEBHOOK_SECRET": SECRET,
        "MAX_WEBHOOK_URL": "https://api.example.org" + HOOK,
        "MAX_SITE_URL": "https://example.org",
    }.items():
        monkeypatch.setenv(key, value)
    get_settings.cache_clear()
    await session.configure(identity_url="sqlite+aiosqlite:///:memory:", clinical_url="sqlite+aiosqlite:///:memory:")
    await init_db()
    async with session.get_clinical_session() as s:
        await repo.upsert_profile(s, token=TOKEN, doctor_id="")
        await repo.set_dialog_state(
            s, token=TOKEN, state="idle", data={"consent": True, "adult": True}, now=clock.now()
        )
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
        yield c
    await session.configure()
    get_settings.cache_clear()
@pytest.mark.parametrize("auth", [{}, {"Authorization": "Bearer unknown"}, {"Authorization": "Basic test"}])
 async def test_auth_required(client, auth):
    assert (await client.get(BASE, headers=auth)).status_code == 401
    assert (await client.post(BASE + "/link", headers=auth, json={"consent": True})).status_code == 401
    assert (await client.delete(BASE, headers=auth)).status_code == 401
 async def test_link_requires_consent_and_hides_patient_credential(client):
    for body in ({}, {"consent": False}, {"consent": True, "user_id": 100}):
        assert (await client.post(BASE + "/link", headers=AUTH, json=body)).status_code == 422
    response = await client.post(BASE + "/link", headers=AUTH, json={"consent": True})
    assert response.status_code == 200
    assert TOKEN not in response.text
    url = urlsplit(response.json()["url"])
    assert url.netloc == "max.ru" and url.path == "/test_bot"
    code = parse_qs(url.query)["start"][0]
    async with session.get_identity_session() as s:
        assert await max_store.bind(s, code, 100, clock.now()) == TOKEN
    assert (await client.get(BASE, headers=AUTH)).json()["connected"] is True
    assert (await client.delete(BASE, headers=AUTH)).json()["connected"] is False
 async def test_revoke_consent_prevents_linking(client):
    async with session.get_clinical_session() as s:
        await repo.set_dialog_state(
            s, token=TOKEN, state="idle", data={"consent": False, "adult": True}, now=clock.now()
        )
    assert (await client.post(BASE + "/link", headers=AUTH, json={"consent": True})).status_code == 403
 async def test_disabled_service_still_allows_disconnect(client, monkeypatch):
    monkeypatch.setenv("MAX_ENABLED", "false")
    get_settings.cache_clear()
    assert (await client.get(BASE, headers=AUTH)).json()["available"] is False
    assert (await client.post(BASE + "/link", headers=AUTH, json={"consent": True})).status_code == 503
    assert (await client.post(HOOK, json={})).status_code == 503
    assert (await client.delete(BASE, headers=AUTH)).status_code == 200
 async def test_webhook_secret_checked_before_body_or_adapter(client):
    assert (await client.post(HOOK, content="not-json")).status_code == 403
    assert (await client.post(HOOK, headers={"X-Max-Bot-Api-Secret": "wrong"}, json={})).status_code == 403
    headers = {"X-Max-Bot-Api-Secret": SECRET}
    assert (await client.post(HOOK, headers=headers, content="not-json")).status_code == 400
    assert (await client.post(HOOK, headers=headers, json=[])).status_code == 422
    assert (await client.post(HOOK, headers=headers, content=" " * 65537)).status_code == 413
 async def test_webhook_storage_failure_is_retryable_without_identity_details(client, monkeypatch):
    handler = AsyncMock(side_effect=OperationalError("private SQL", {"token": TOKEN}, Exception("private")))
    monkeypatch.setitem(sys.modules, "app.channels.max_bot", SimpleNamespace(handle_update=handler))
    response = await client.post(HOOK, headers={"X-Max-Bot-Api-Secret": SECRET}, json={})
    assert response.status_code == 503 and TOKEN not in response.text and "private" not in response.text
@@ -53,15 +53,7 @@ async def _tables(engine_fn: Callable[[], AsyncEngine]) -> set[str]:
 async def test_create_all_puts_tables_in_the_right_db(dbs: None) -> None:
    identity_tables = await _tables(session.identity_engine)
    clinical_tables = await _tables(session.clinical_engine)
-    assert identity_tables == {
+    assert identity_tables == {"patients", "doctors", "connect_codes"}
        "patients",
        "doctors",
        "connect_codes",
        "max_bindings",
        "max_link_codes",
        "max_webhook_receipts",
        "max_deliveries",
    }
    assert {
        "patient_profiles",
        "item_responses",
@@ -1,123 +0,0 @@
 """Persistent, one-time linking and queue recovery; synthetic identities only."""
 from datetime import timedelta
 import pytest
 from sqlalchemy import select
 from app.core import clock
 from app.db import repo, session
 from app.db.bootstrap import init_db
 from app.db.clinical import ScaleResult
 from app.db.max_notifications import MaxBinding, MaxDelivery, MaxLinkCode
 from app.integrations import max_store as store
@pytest.fixture(autouse=True)
 async def database():
    await session.configure(identity_url="sqlite+aiosqlite:///:memory:", clinical_url="sqlite+aiosqlite:///:memory:")
    await init_db()
    yield
    await session.configure()
 async def test_codes_are_hashed_expiring_one_time_and_replaced():
    now = clock.now()
    async with session.get_identity_session() as s:
        old, _ = await store.issue_code(s, "patient", now, 60)
        code, expires = await store.issue_code(s, "patient", now, 60)
        assert await store.code_token(s, old, now) is None
        row = (await s.scalars(select(MaxLinkCode))).one()
        assert row.digest != code and row.digest == store.digest(code)
        assert await store.code_token(s, code, expires) is None
        assert await store.bind(s, code, 100, now) == "patient"
        assert await store.bind(s, code, 100, now) is None
    async with session.get_identity_session() as s:
        assert (await s.get(MaxBinding, "patient")).user_id == 100
 async def test_active_binding_cannot_be_reassigned_in_either_direction():
    now = clock.now()
    async with session.get_identity_session() as s:
        code, _ = await store.issue_code(s, "one", now, 60)
        await store.bind(s, code, 100, now)
        code, _ = await store.issue_code(s, "two", now, 60)
        assert await store.bind(s, code, 100, now) is None
        assert await store.code_token(s, code, now) == "two"
        code, _ = await store.issue_code(s, "one", now, 60)
        assert await store.bind(s, code, 200, now) is None
        await store.disconnect(s, "one")
        assert await store.code_token(s, code, now) is None
        code, _ = await store.issue_code(s, "two", now, 60)
        assert await store.bind(s, code, 100, now) == "two"
 async def test_disconnect_cancels_pending_delivery_and_prevents_stale_ack():
    now = clock.now()
    async with session.get_identity_session() as s:
        await store.enqueue(s, key="job", user_id=100, token="one", kind="notification", now=now)
        row = await store.claim(s, now)
    async with session.get_identity_session() as s:
        await store.disconnect(s, "one")
        await store.finish(s, row, state="sent", now=now)
    async with session.get_identity_session() as s:
        assert (await s.get(MaxDelivery, "job")).state == "cancelled"
 async def test_queue_survives_sessions_deduplicates_and_recovers_expired_lease():
    now = clock.now()
    async with session.get_identity_session() as s:
        for _ in range(2):
            await store.enqueue(s, key="job", user_id=100, kind="help", now=now)
    async with session.get_identity_session() as s:
        first = await store.claim(s, now)
        assert first.attempts == 1
    async with session.get_identity_session() as s:
        assert await store.claim(s, now + timedelta(seconds=59)) is None
    async with session.get_identity_session() as s:
        second = await store.claim(s, now + timedelta(seconds=61))
        assert second.id == first.id and second.lease != first.lease and second.attempts == 2
    async with session.get_identity_session() as s:
        await store.finish(s, first, state="sent", now=now)
    async with session.get_identity_session() as s:
        assert (await s.get(MaxDelivery, "job")).state == "sending"
        await store.finish(s, second, state="sent", now=now)
    async with session.get_identity_session() as s:
        assert await store.claim(s, now + timedelta(days=1)) is None
 async def test_retry_deadline_and_attempt_limit():
    now = clock.now()
    async with session.get_identity_session() as s:
        await store.enqueue(s, key="job", user_id=100, kind="help", now=now)
    for attempt in range(store.MAX_ATTEMPTS):
        when = now + timedelta(seconds=attempt * 120)
        async with session.get_identity_session() as s:
            row = await store.claim(s, when)
        async with session.get_identity_session() as s:
            await store.finish(s, row, state="queued", now=when, retry_seconds=120, error="http_429")
        async with session.get_identity_session() as s:
            assert await store.claim(s, when + timedelta(seconds=119)) is None
    async with session.get_identity_session() as s:
        assert await store.claim(s, now + timedelta(days=1)) is None
    async with session.get_identity_session() as s:
        row = await s.get(MaxDelivery, "job")
        assert row.state == "failed" and row.attempts == store.MAX_ATTEMPTS
 async def test_completion_scan_ignores_old_and_incomplete_tests_and_recovers_without_duplicates():
    now = clock.now()
    async with session.get_clinical_session() as s:
        await repo.upsert_profile(s, token="one", doctor_id="")
        await repo.set_dialog_state(s, token="one", state="idle", data={"consent": True, "adult": True}, now=now)
        for ts, complete in ((now - timedelta(days=1), True), (now, True), (now, False)):
            s.add(ScaleResult(token="one", ts=ts, scale="phq9", total=0, severity="none", complete=complete))
    async with session.get_identity_session() as s:
        s.add(MaxBinding(token="one", user_id=100, enabled=True, connected_at=now))
    for _ in range(2):
        async with session.get_identity_session() as s:
            await store.collect_completed(s, now)
    async with session.get_identity_session() as s:
        row = (await s.scalars(select(MaxDelivery))).one()
        assert row.kind == "completed" and row.token == "one"
        assert "phq9" not in row.id
@@ -197,75 +197,3 @@ async def test_empty_final_answer_is_nudged(sertraline):
    assert ans is not None and "Алкоголь" in ans.text
    assert sent[0]["tool_choice"] == "required" and "tool_choice" not in sent[1]
    assert "сертралин" in sent[0]["messages"][0]["content"]  # patient brief in the system prompt
 async def test_region_block_string_error_degrades():
    # Seen live from the RF server: 403 {"success": false, "error": "Access denied by security policy."}
    client = httpx.AsyncClient(
        transport=httpx.MockTransport(lambda r: httpx.Response(403, json={"success": False, "error": "Access denied"}))
    )
    provider = OpenRouterProvider(_settings(), client=client)
    assert await agent.answer_chat(ANDREY, scrub("вопрос"), provider=provider, settings=_settings()) is None
 async def test_unexpected_provider_bug_degrades():
    class Broken:
        name = "broken"
        async def chat(self, messages, tools):
            raise AttributeError("boom")
    assert await agent.answer_chat(ANDREY, scrub("вопрос"), provider=Broken(), settings=_settings()) is None
 # ------------------------------------------------------------------------------------------------- gigachat
 class _FakeGiga:
    """Mimics gigachat.GigaChat.achat: first a function_call, then the final text."""
    def __init__(self):
        self.chats = []
    async def achat(self, chat):
        from types import SimpleNamespace as NS
        self.chats.append(chat)
        if len(self.chats) == 1:
            fc = NS(name="check_drug_for_patient", arguments={"drug": "ибупрофен"})
            msg = NS(content="", function_call=fc)
        else:
            msg = NS(
                content="Ибупрофен вместе с сертралином повышает риск кровотечений. Обсудите с врачом.",
                function_call=None,
            )
        return NS(choices=[NS(message=msg)], usage=NS(total_tokens=42))
 async def test_gigachat_tool_loop(sertraline):
    from app.llm.agent.providers import GigaChatToolProvider
    fake = _FakeGiga()
    s = Settings(_env_file=None, chat_agent="gigachat", gigachat_credentials="x", chat_agent_timeout_s=5.0)
    ans = await agent.answer_chat(
        ANDREY, scrub("Можно ибупрофен?"), provider=GigaChatToolProvider(s, client=fake), settings=s
    )
    assert ans is not None and "кровотеч" in ans.text and ans.provider == "gigachat"
    second = fake.chats[1]
    roles = [str(m.role.value if hasattr(m.role, "value") else m.role) for m in second.messages]
    assert roles == ["system", "user", "assistant", "function"]
    assert second.messages[2].function_call.name == "check_drug_for_patient"
    assert second.messages[3].name == "check_drug_for_patient"
    assert {f.name for f in second.functions} == {
        "search_guidelines",
        "get_patient_record",
        "get_drug_info",
        "check_drug_for_patient",
    }
 def test_build_provider_prefers_gigachat():
    from app.llm.agent.providers import GigaChatToolProvider
    s = Settings(_env_file=None, chat_agent="auto", gigachat_credentials="x", openrouter_api_key="y")
    assert isinstance(agent.build_provider(s), GigaChatToolProvider)