diff --git a/.env.example b/.env.example index 9b90399..43c7fa1 100644 --- a/.env.example +++ b/.env.example @@ -23,6 +23,8 @@ CLINICAL_DB_URL=postgresql+asyncpg://clinical_app:clinical_app@localhost:5432/cl # --- Telegram: prototype on synthetic data only; prod channel is MAX / Gosuslugi. Empty = bot disabled. TG_BOT_TOKEN= +# http://user:pass@host:port or socks5://… — needed on the RF server (docker has no route to Telegram over IPv4) +TG_PROXY_URL= DOCTOR_NOTIFY_CHAT_ID= # --- Demo virtual clock (ADR-005) diff --git a/Makefile b/Makefile index 6e871cc..a379f33 100644 --- a/Makefile +++ b/Makefile @@ -6,7 +6,7 @@ install: up: @test -f .env || cp .env.example .env docker compose up -d --build - @echo "API: http://localhost:8000/docs Dashboard: http://localhost:8501" + @echo "Patient app: http://localhost:8080 API: http://localhost:8000/docs Doctor/demo console: http://localhost:8501" down: docker compose down diff --git a/README.md b/README.md index 09cf25f..8ec9919 100644 --- a/README.md +++ b/README.md @@ -1,14 +1,117 @@ # PTSRaiAgent — ИИ-помощник сопровождения при ПТСР и депрессии -> Хакатон MedITron · кейс SberMed AI «ИИ-помощник пациента». Прототип на **синтетических данных**. -> Telegram используется только для прототипа на синтетических данных; прод-канал — MAX / Госуслуги. +> Хакатон MedITron · кейс SberMed AI «ИИ-помощник пациента». Прототип работает **только на синтетических данных**. +> Демо-стенд: **https://meditron.tenvex.ru** (кабинет врача: https://meditron.tenvex.ru/doctor/, API: https://meditron.tenvex.ru/docs). -🚧 README заполняется к сдаче (issue «README по требованиям оформления»). Разделы по требованиям жюри: +--- -1. Цели и задачи решения -2. Что реализовано -3. Развёртывание, запуск, использование -4. Входные и выходные данные, API -5. Ограничения и особенности медицинской тематики +## 1. Цели и задачи -Пока смотрите [`CLAUDE.md`](CLAUDE.md) и [`docs/`](docs/). +**Проблема.** Ветераны СВО и люди с депрессией до помощи доходят с трудом, а начав лечение, часто его бросают: в первые недели побочные эффекты уже есть, а эффекта ещё нет. Врач между визитами ничего не знает о пациенте. + +**Решение.** Цифровой помощник сопровождает пациента между визитами к врачу. Из сценариев кейса закрыты два: **мониторинг после обращения** и **управление приёмом лекарств**. +- Пациент общается с помощником в чате: веб-приложение или Telegram. В проде канал — MAX / Госуслуги. +- Помощник собирает валидированные шкалы, ежедневный чек-ин и отметки о приёме препарата. +- Он отвечает на вопросы о лечении по клиническим рекомендациям Минздрава, по инструкции ГРЛС и по карте пациента в ЕМИАС. +- Врач видит «красные зоны» и получает сигналы при ухудшении. + +**Принципы.** +- ИИ не ставит диагнозы и не назначает лечение. +- LLM только формулирует ответы и извлекает данные. +- **Риск, алерты и скоринг считают детерминированные правила** по валидированным шкалам (PHQ-9, PCL-5, GAD-7, AUDIT-C, C-SSRS, DAR-5). +- Клинические решения принимает врач. + +## 2. Что реализовано + +| Блок | Что есть | +|---|---| +| **Чат-агент** | LLM с инструментами только для чтения: поиск по базе клинических рекомендаций (RAG, депрессия 2024, ПТСР 2023), карта пациента из ЕМИАС (мок), инструкция ГРЛС, детерминированная проверка препарата под пациента (сопутствующие болезни, непереносимость вспомогательных веществ, взаимодействия). Ответ печатается в реальном времени (SSE), видны шаги агента. Провайдеры: GigaChat (функции), OpenRouter (временно, только синтетика), офлайн-mock. | +| **Диалоговый движок** | Обезличивание → проверка безопасности → FSM. Согласие (152-ФЗ, отзыв в один шаг), проверка возраста, онбординг ветерана (PC-PTSD-5 → PCL-5 → PHQ-9 → AUDIT-C), опросник intake (C-SSRS, PHQ-2→9, TSQ→PCL-5, RUS-AUDIT-S), ежедневный чек-ин, напоминания о дозах, побочные эффекты. На вопросы отвечает в любом состоянии. | +| **Безопасность** | Лексикон риска (100% recall на наборе кризисных фраз), фиксированный кризисный скрипт без LLM, алерт врачу и дежурному, план безопасности. В демо кризисное окно показывается в конце теста и не блокирует диалог (`CRISIS_BLOCKING`). | +| **Правила мониторинга** | 17 правил в `config/rules.yaml`: п.9 PHQ-9, рост баллов, побочки 3 дня подряд, намерение бросить, пропуски 3 из 7, сон < 4 ч, алкоголь, молчание > 3 дней и другие. | +| **Кабинет врача** | Список «красных зон» с индексом риска прерывания терапии, карточка (графики шкал, сон, настроение, дозы, алерты, план безопасности, ИИ-саммари), живые алерты, подключение пациента с кодом и QR, аудит доступа. | +| **Веб-приложение пациента** | Вход через ЕМИАС (демо: код от врача), анонимный вход или демо-вход в один клик (у каждого посетителя своя копия пациента). Экраны: сегодня, лекарства, чат, врач, профиль. | +| **Демо-консоль** | Виртуальные часы (13 дней терапии за 3 минуты), проигрывание сценария, сплит-экран, тумблер начального теста. | +| **Инфраструктура** | Docker compose, CI (ruff + pytest, около 1060 тестов), CD: мерж в `dev` выкатывает стенд. | + +## 3. Развёртывание и запуск + +**Требования:** только Docker с Compose v2 — всё остальное собирается в контейнерах. Для разработки — Python 3.12 и [uv](https://docs.astral.sh/uv/), Node 20. + +```bash +cp .env.example .env # по умолчанию офлайн: LLM_PROVIDER=mock, агент выключен +make up # postgres + API (:8000) + кабинет врача/демо-консоль (:8501) + приложение пациента (:8080) +make seed # синтетический кабинет: врач, ветеран Андрей и 7 пациентов +``` + +- API и Swagger: http://localhost:8000/docs +- Кабинет врача и демо-консоль: http://localhost:8501 +- Веб-приложение пациента: http://localhost:8080 (собирается внутри Docker, Node на машине не нужен; для разработки — `cd web && npm run dev`, :5173) + +**Включить чат-агента** в `.env`: +- `CHAT_AGENT=gigachat` + `GIGACHAT_CREDENTIALS=…` — целевой вариант для РФ; +- или `CHAT_AGENT=openrouter` + `OPENROUTER_API_KEY=…` (+ `OPENROUTER_PROXY_URL` с сервера в РФ) — только для прототипа; +- проверить из терминала: `uv run python -m app.llm.agent "Можно мне ибупрофен?" --persona andrey`. + +**Демо-переключатели:** +- `SCREENING_ENABLED` — начальный тест (дублируется тумблером в демо-консоли); +- `CRISIS_BLOCKING` — клинический протокол: кризис блокирует диалог до подтверждения врача; +- `DEMO_CLOCK` — виртуальные часы; +- `INTAKE_DEMO` — опросники intake на синтетике. + +**Сервер:** `infra/deploy/` — override compose, nginx-vhost, `deploy.sh`. Выкатка на стенд — `.gitea/workflows/deploy.yml`. + +**Разработка:** `uv sync`, `make lint test`. + +## 4. Входные и выходные данные, API + +**Входные данные:** +- сообщения и ответы кнопками пациента; +- назначения (мок МИС, `data/meds/mis_prescriptions.json`); +- карта ЕМИАС (мок, `data/emias/records.yaml`); +- выдержки ГРЛС (`data/meds/grls/`); +- клинические рекомендации (`config/rag/corpus/`). + +**Выходные данные:** ответы пациенту, баллы шкал, алерты врачу, карточка пациента, журнал аудита. + +Все клинические конфиги (шкалы, пороги, правила, тексты, лексикон) лежат в `config/` и правятся без кода. + +Основные эндпоинты (`/api/v1`, полная спецификация — `/docs`): + +| Метод и путь | Назначение | +|---|---| +| `POST /dialog/{token}/events` | событие диалога: текст, кнопка, start → ответы помощника | +| `POST /dialog/{token}/events/stream` | то же в виде SSE: шаги агента, текст по мере генерации, финальные ответы | +| `GET /dialog/{token}/outbox` | системные сообщения: напоминания, чек-ины | +| `POST /auth/emias`, `/auth/emias/demo`, `/auth/demo/fresh`, `/auth/anonymous` | входы веб-приложения | +| `GET /me/{token}/today` | план на сегодня: дозы, отметки за неделю, визит | +| `GET /doctors/{id}/patients` | список пациентов врача с цветом и индексом риска | +| `GET /patients/{token}?doctor_id=` | карточка пациента (пишется в аудит) | +| `GET /doctors/{id}/alerts`, `POST /alerts/{id}/ack` | алерты и их подтверждение | +| `POST /patients/connect` | подключение пациента: назначение и одноразовый код | +| `GET/POST /demo/clock…`, `/demo/reset`, `/demo/settings` | демо-управление | + +## 5. Ограничения и особенности медицинской тематики + +- **Только синтетические данные.** Персоны, карты ЕМИАС, назначения и выдержки ГРЛС вымышлены или пересказаны для прототипа и подлежат сверке. +- **Не медицинское изделие в текущем виде.** Сервис не ставит диагнозы и не меняет терапию. Пороги шкал, тексты и лексикон — стартовые значения из литературы, их утверждает психиатр (`docs/clinical/`). +- **152-ФЗ:** отдельное согласие на обработку данных о здоровье, две раздельные БД (identity и clinical) со связью через HMAC-токен, в LLM уходит только обезличенный текст, каждое обращение врача пишется в аудит. +- **Зарубежные сервисы.** OpenRouter используется временно и только на синтетике. Для реальных пациентов нужен GigaChat в РФ-контуре. +- **Безопасность.** Кризисный ответ фиксированный и выдаётся без LLM. Пропуск кризиса недопустим, ложные срабатывания допустимы. Демо-режим (кризис без блокировки) не равен клиническому протоколу: в проде `CRISIS_BLOCKING=true`. +- **Ответы LLM** могут слегка различаться от раза к разу. Фактура берётся только из инструментов, а запрещённые формулировки (диагнозы, дозы, «отмените») фильтруются. + +## Структура репозитория + +``` +app/ бэкенд: core (privacy, safety, шкалы, правила, часы), dialog (движок и сценарии), llm (шлюз, RAG, + чат-агент), db, api, scheduler, integrations (моки МИС/ЕМИАС/ГРЛС, уведомления), services +dashboard/ кабинет врача и демо-консоль (Streamlit) +web/ веб-приложение пациента (React + Vite) +config/ клинические конфиги: шкалы, правила, модули, тексты, лексикон, RAG-корпус +data/ синтетика: сценарий демо, карты ЕМИАС, назначения, выдержки ГРЛС +infra/ postgres init, деплой на сервер +tests/ pytest (около 1060 тестов) +docs/ спецификация, ADR (DECISIONS.md), клинические материалы, концепция и CJM, регламент +``` + +Подробнее: [`docs/info/SPEC.md`](docs/info/SPEC.md) · [`docs/DECISIONS.md`](docs/DECISIONS.md) · [`docs/info/CLINICAL_MODULES.md`](docs/info/CLINICAL_MODULES.md) · [`docs/source/`](docs/source/) (кейс, концепция, CJM). diff --git a/app/api/fixtures.py b/app/api/fixtures.py index 9b9904c..eae1c58 100644 --- a/app/api/fixtures.py +++ b/app/api/fixtures.py @@ -25,7 +25,8 @@ DOCTOR_ID = "doc-smirnova" # Synthetic personas for the demo console and the demo ЕМИАС sign-in of the patient web app. DEMO_PERSONAS: list[DemoPersona] = [ - DemoPersona(token="tok-andrey", display_name="Андрей С.", therapy_day=13, scenario="data/synthetic/andrey.jsonl"), + # ЕМИАС demo sign-in shows this persona as «Никита В.» (the one-click demo on the welcome screen is «Андрей С.»). + DemoPersona(token="tok-andrey", display_name="Никита В.", therapy_day=13, scenario="data/synthetic/andrey.jsonl"), ] diff --git a/app/api/patient_web.py b/app/api/patient_web.py index daa3231..d5907ef 100644 --- a/app/api/patient_web.py +++ b/app/api/patient_web.py @@ -158,6 +158,7 @@ async def auth_emias_demo(body: EmiasDemoIn) -> AuthOut: DEMO_SOURCE = "tok-andrey" +ONE_CLICK_NAME = "Андрей С." # «Демо: войти как Андрей»; the ЕМИАС demo persona is named in fixtures (Никита) @router.post("/auth/demo/fresh", response_model=AuthOut) @@ -172,12 +173,13 @@ async def _fresh_demo_copy(*, start_in_chat: bool) -> AuthOut: token = f"{DEMO_SOURCE}-{secrets.token_hex(4)}" now = clock.now() persona = next(p for p in fixtures.DEMO_PERSONAS if p.token == DEMO_SOURCE) + name = persona.display_name if not start_in_chat else ONE_CLICK_NAME snils = f"000-{secrets.randbelow(10**3):03d}-{secrets.randbelow(10**3):03d} {secrets.randbelow(100):02d}" async with get_identity_session() as s: await identity_repo.create_patient( s, token=token, - display_name=f"{persona.display_name} (демо)", + display_name=f"{name} (демо)", snils=snils, clinic_id="demo", doctor_id=fixtures.DOCTOR_ID, @@ -194,7 +196,7 @@ async def _fresh_demo_copy(*, start_in_chat: bool) -> AuthOut: consent = {"consent": True, "consent_ver": "demo", "consent_at": now.isoformat(), "revoked_at": None} await repo.set_dialog_state(s, token=token, state="idle", data={**consent, "adult": True}, now=now) log.info("web auth: demo fresh tok=%s", token[:14]) - return AuthOut(token=token, mode="emias", display_name=persona.display_name) + return AuthOut(token=token, mode="emias", display_name=name) @router.post("/auth/anonymous", response_model=AuthOut) diff --git a/app/api/routes.py b/app/api/routes.py index b6fb171..44a7053 100644 --- a/app/api/routes.py +++ b/app/api/routes.py @@ -8,9 +8,14 @@ disclosed — the dashboard polls every 2–3 s), `ack_alert` (in `repo.ack_aler `app/api/fixtures.py` is kept for tests and as the demo cabinet reference (doctor id). """ +import asyncio +import json +import logging import secrets +from typing import Any from fastapi import APIRouter, HTTPException +from fastapi.responses import StreamingResponse from sqlalchemy import select from app.api import demo as console @@ -42,10 +47,12 @@ from app.db.identity import Patient from app.db.session import get_clinical_session, get_identity_session from app.dialog.contracts import Event, get_engine from app.integrations.base import integrations +from app.llm.agent import stream as agent_stream from app.scheduler import jobs as scheduler from app.services.anonymous_care import display_name as anonymous_name from app.services.monitoring import refresh_status +log = logging.getLogger(__name__) router = APIRouter(prefix="/api/v1") router.include_router(max_notifications.router) router.include_router(questionnaire_router) @@ -219,6 +226,49 @@ async def dialog_event(token: str, body: DialogIn) -> DialogOut: return DialogOut(replies=await get_engine().handle(token, body)) +@router.post("/dialog/{token}/events/stream", tags=["dialog"]) +async def dialog_event_stream(token: str, body: DialogIn) -> StreamingResponse: + """Same as /events, as Server-Sent Events: `status` (agent steps), `delta` (answer as it is written), + `reset`, then `done` with the final replies (identical to /events). `error` if the engine failed.""" + await _require_patient(token) + queue: asyncio.Queue[tuple[str, Any]] = asyncio.Queue() + marker = agent_stream.STREAM.set(queue) # copied into the task's context below + try: + + async def run() -> None: + try: + replies = await get_engine().handle(token, body) + queue.put_nowait(("done", [r.model_dump() for r in replies])) + except Exception: # noqa: BLE001 — reported to the client, which falls back to /events + log.exception("dialog stream failed") + queue.put_nowait(("error", None)) + + task = asyncio.create_task(run()) + finally: + agent_stream.STREAM.reset(marker) + + async def events(): + try: + while True: + try: + kind, payload = await asyncio.wait_for(queue.get(), timeout=15) + except TimeoutError: + yield ": keepalive\n\n" + continue + yield f"event: {kind}\ndata: {json.dumps(payload, ensure_ascii=False)}\n\n" + if kind in ("done", "error"): + return + finally: + if not task.done(): + await task + + return StreamingResponse( + events(), + media_type="text/event-stream", + headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}, # nginx: do not buffer SSE + ) + + @router.get("/dialog/{token}/outbox", response_model=DialogOut, tags=["dialog"]) async def dialog_outbox(token: str) -> DialogOut: """System-initiated replies (reminders, check-ins) waiting for the web console.""" diff --git a/app/channels/telegram.py b/app/channels/telegram.py index 486d36a..66cda0f 100644 --- a/app/channels/telegram.py +++ b/app/channels/telegram.py @@ -19,6 +19,7 @@ from typing import Protocol import yaml from aiogram import Bot, Dispatcher, F, Router +from aiogram.client.session.aiohttp import AiohttpSession from aiogram.filters import CommandObject, CommandStart from aiogram.types import CallbackQuery, InlineKeyboardButton, InlineKeyboardMarkup, Message from pydantic import BaseModel @@ -193,11 +194,28 @@ class TelegramChannel: await self._dispatch(chat_id, person.token, Event(type="start", params={"code": code})) return token = await self.directory.token_for(chat_id) + if token is None and get_settings().demo_clock: + token = await self._demo_copy(chat_id) if token is None: await self._say(chat_id, texts.not_connected) return await self._dispatch(chat_id, token, Event(type="start")) + async def _demo_copy(self, chat_id: int) -> str | None: + """Demo stand (DEMO_CLOCK): `/start` without a code gets a private copy of veteran Andrey (synthetic), + starting from consent and the intake — the same entry as the ЕМИАС demo button of the web app.""" + from app.api.patient_web import _fresh_demo_copy + + try: + token = (await _fresh_demo_copy(start_in_chat=False)).token + except Exception: + log.exception("start: demo copy failed") + return None + if not await self.directory.bind(chat_id, token): + return None + log.info("start: demo copy bound tok=%s", _short(token)) + return token + async def on_text(self, chat_id: int, text: str | None) -> None: texts = load_texts() token = await self.directory.token_for(chat_id) @@ -272,6 +290,12 @@ _task: asyncio.Task[None] | None = None _prev_outbox: Outbox | None = None +def make_bot(token: str) -> Bot: + """aiogram `Bot`, through `TG_PROXY_URL` when it is set.""" + proxy = get_settings().tg_proxy_url + return Bot(token, session=AiohttpSession(proxy=proxy)) if proxy else Bot(token) + + async def start() -> None: """Start polling if TG_BOT_TOKEN is set; no-op otherwise.""" global _bot, _dp, _task, _prev_outbox @@ -280,7 +304,7 @@ async def start() -> None: if not token: log.info("telegram disabled: TG_BOT_TOKEN is empty") return - bot = Bot(token) + bot = make_bot(token) channel = TelegramChannel(bot) dp = Dispatcher() dp.include_router(build_router(channel)) diff --git a/app/config.py b/app/config.py index 40b62bf..27e5cb6 100644 --- a/app/config.py +++ b/app/config.py @@ -44,6 +44,8 @@ class Settings(BaseSettings): # Telegram is used only for the prototype on synthetic data; prod channel is MAX / Gosuslugi. tg_bot_token: str = "" + # The RF server reaches api.telegram.org over IPv6 only; docker has IPv4 only → go through a proxy (http/socks). + tg_proxy_url: str = "" doctor_notify_chat_id: int | None = None # MAX is opt-in. Its adapter is delivered in the related backend/channel changes. @@ -80,6 +82,9 @@ class Settings(BaseSettings): demo_clock: bool = True # Initial screening (onboarding questionnaires) on/off — the default; the demo console can flip it at runtime. screening_enabled: bool = True + # Crisis blocks the dialog until the doctor's ack (clinical protocol) — off for the demo: the crisis message + # and the doctor's alert still happen, the patient keeps using everything. + crisis_blocking: bool = False tz: str = "Europe/Moscow" api_base_url: str = "http://localhost:8000" diff --git a/app/core/intake_questions.py b/app/core/intake_questions.py index 694ac68..c4913c6 100644 --- a/app/core/intake_questions.py +++ b/app/core/intake_questions.py @@ -61,6 +61,11 @@ def red_flag_category(item_id: str) -> str | None: return next((i["category"] for i in red_flags()["items"] if i["id"] == item_id), None) +def red_flag_triggers(item_id: str) -> bool: + """False for questions that are asked and stored but never stop/alert (rf_weapons, team decision 2026-10-04).""" + return next((i.get("trigger", True) for i in red_flags()["items"] if i["id"] == item_id), True) + + def next_intake(answers: dict, policy: IntakePolicy, sex: str | None) -> Question | None: """Stage 1: the red-flag block first (config/instruments/red_flags.yaml), then the fixed order of «Опросники для MVP»: PHQ-2, trauma filter → TSQ, RUS-AUDIT-S, C-SSRS screener, the context questions; stage 2 (rest of diff --git a/app/dialog/crisis.py b/app/dialog/crisis.py index 9eaa4bd..e82b32f 100644 --- a/app/dialog/crisis.py +++ b/app/dialog/crisis.py @@ -14,6 +14,7 @@ Decisions: from typing import Any +from app.config import get_settings from app.core.safety import CrisisResource, CrisisScript, crisis_reply from app.dialog.contracts import Reply from app.dialog.fsm import Ctx, Step @@ -37,6 +38,9 @@ def open_crisis(ctx: Ctx, categories: list[str], source: str) -> Step: text = "\n".join([ctx.texts.crisis.repeat, *_resource_lines(script.resources)]) return Step(replies=[Reply(text=text, kind="crisis")]) crisis = {"open": True, "opened_at": ctx.now.isoformat(), "source": source, "variant": script.variant} + if not get_settings().crisis_blocking: + # Non-blocking (demo default): the script is shown, the doctor is alerted, the patient keeps going. + return Step(replies=[Reply(text=script.text, kind="crisis")], data={"crisis": crisis}) return Step( replies=[Reply(text=script.text, kind="crisis")], next_state="crisis", diff --git a/app/dialog/engine.py b/app/dialog/engine.py index 2f7d43e..724022a 100644 --- a/app/dialog/engine.py +++ b/app/dialog/engine.py @@ -37,6 +37,7 @@ from weakref import WeakValueDictionary from sqlalchemy.exc import InterfaceError, OperationalError from sqlalchemy.ext.asyncio import AsyncSession +from app.config import get_settings from app.core import clock, safety from app.core.intake_runtime import intake_enabled from app.core.privacy import scrub @@ -68,6 +69,9 @@ from app.services.contracts import MonitoringService, get_monitoring log = logging.getLogger(__name__) LAST_PROMPT = "last_prompt" +DEFERRED_CRISIS = "deferred_crisis" +QUESTIONNAIRE_PREFIXES = ("scale.", "veteran.onboarding.", "intake.") +DEFERRABLE_KINDS = frozenset({"crisis", "safety_plan"}) MEDICINE_WORDS = re.compile( r"(лекарств|таблет|препарат|сертралин|пароксетин|флуоксетин|эсциталопрам|антидепрессант|побочк)", re.I ) @@ -145,6 +149,8 @@ class Engine: await self._apply(ctx, await self._start_pending(ctx), run) + self._defer_crisis_messages(ctx, event, run) + if _consented(ctx): for reply in run.replies: # Outgoing texts may quote the safety plan (names of relatives) — scrub before storing too. @@ -177,6 +183,21 @@ class Engine: ) return run.replies + def _defer_crisis_messages(self, ctx: Ctx, event: Event, run: _Run) -> None: + """Crisis messages triggered by test answers wait until the questionnaire is over (team decision + 2026-10-04: let the patient finish the test, then show the crisis message; it blocks nothing). The doctor's + alert is not delayed — only the patient-facing message. A typed acute message is never deferred.""" + if get_settings().crisis_blocking: + return # clinical protocol: the crisis takes over immediately + if event.type == "button" and ctx.state.startswith(QUESTIONNAIRE_PREFIXES): + held = [r for r in run.replies if r.kind in DEFERRABLE_KINDS] + if held: + run.replies[:] = [r for r in run.replies if r.kind not in DEFERRABLE_KINDS] + ctx.data[DEFERRED_CRISIS] = [*(ctx.data.get(DEFERRED_CRISIS) or []), *(r.model_dump() for r in held)] + return + if ctx.data.get(DEFERRED_CRISIS) and not ctx.state.startswith(QUESTIONNAIRE_PREFIXES): + run.replies.extend(Reply.model_validate(r) for r in ctx.data.pop(DEFERRED_CRISIS)) + def _abandoned(self, ctx: Ctx) -> bool: """The current question step was left unanswered longer than `abandon_after_hours` (app.core.clock).""" if intake_enabled() or not _consented(ctx) or ctx.state in RESTING_STATES or crisis_open(ctx.data): diff --git a/app/dialog/flows/intake.py b/app/dialog/flows/intake.py index e5479f3..04acede 100644 --- a/app/dialog/flows/intake.py +++ b/app/dialog/flows/intake.py @@ -4,6 +4,7 @@ import math from dataclasses import asdict, replace from zoneinfo import ZoneInfo +from app.config import get_settings from app.core import interpretation from app.core.intake_questions import ( DIARY_KINDS, @@ -15,6 +16,7 @@ from app.core.intake_questions import ( group_answers, next_intake, red_flag_category, + red_flag_triggers, scale_question, texts, ) @@ -116,6 +118,39 @@ async def urgent(ctx, row=None, *, code="help_now", categories=None): return step +def _defer_urgent(row, code, categories=None): + """Remember the first crisis signal of this questionnaire; the questionnaire goes on (team decision + 2026-10-04: the crisis message comes at the end and blocks nothing).""" + if not row.details.get("pending_urgent"): + row.details = {**row.details, "pending_urgent": {"code": code, "categories": categories or []}} + + +async def _flush_urgent(ctx, row, step): + """Questionnaire over: acute alert + callback for the doctor, then the (non-blocking) crisis message.""" + pending = (row.details or {}).get("pending_urgent") + if not pending or step.next_state == "intake.collect": + return step + row.details = {**row.details, "pending_urgent": None} + code, categories = pending["code"], pending.get("categories") or [] + profile = await ctx.profile() + if profile: + profile.enhanced_monitoring = True + await service.alert( + ctx.session, + ctx.token, + f"{row.id}:{code}", + code, + ctx.now, + level=AlertLevel.ACUTE, + params={"record_id": row.id, "deferred": True}, + ) + await schedule_post_crisis_call(ctx.session, ctx.token, ctx.now, source="questionnaire:" + code) + crisis = open_crisis(ctx, categories, source="questionnaire:" + code) + for reply in crisis.replies: + reply.buttons.append([Button(label=ctx.texts.safety_plan.title.rstrip(":"), data="safety_plan")]) + return step.then(crisis) + + async def item9_discrepancy(ctx, row): profile = await ctx.profile() if profile: @@ -170,18 +205,27 @@ async def manual_review(ctx, row, code): async def advance(ctx, row): + return await _flush_urgent(ctx, row, await _advance(ctx, row)) + + +async def _advance(ctx, row): a = row.answers c = score_cssrs_recent(cssrs_answers(a)) policy = load_policy() # Red flags (asked first): any «Да» stops the questionnaire with the help screen of its category. for item_id, value in a.items(): - if item_id.startswith("rf_") and value == 1: + if item_id.startswith("rf_") and value == 1 and red_flag_triggers(item_id): category = red_flag_category(item_id) - return await urgent(ctx, row, code="red_flag_" + item_id[3:], categories=[category] if category else []) + if get_settings().crisis_blocking: # clinical protocol: stop at once + return await urgent(ctx, row, code="red_flag_" + item_id[3:], categories=[category] if category else []) + _defer_urgent(row, "red_flag_" + item_id[3:], [category] if category else []) + break # One crisis rule for the whole app (ADR-013): the crisis screen opens at a HIGH C-SSRS risk only. Moderate → # stop + psychiatrist (about 24 h); low → flag, safety plan, continue. if c.risk_lower_bound == "high": - return await urgent(ctx, row, code="suicide_signal") + if get_settings().crisis_blocking: + return await urgent(ctx, row, code="suicide_signal") + _defer_urgent(row, "suicide_signal") if a.get("dphq_9", 0) > 0 and row.kind in DIARY_KINDS: # Diary depression module, item 9 ≥1 → the C-SSRS right away («Этап 2»); its risk level decides. return await item9_assessment(ctx, row) diff --git a/app/dialog/scale_session.py b/app/dialog/scale_session.py index 48c3657..3111ebf 100644 --- a/app/dialog/scale_session.py +++ b/app/dialog/scale_session.py @@ -33,6 +33,7 @@ from dataclasses import replace from datetime import timedelta from typing import Any +from app.config import get_settings from app.core.privacy import scrub from app.core.scales import Item, ScaleDef, ScaleResult, load_scale, score from app.db import repo @@ -237,7 +238,10 @@ async def _record( sex = profile.sex if profile is not None and profile.sex in ("m", "f") else None result = score(scale.id, answers, sex=sex) # type: ignore[arg-type] crisis_flags = [f for f in result.flags if f in ctx.texts.settings.crisis_flags] - if crisis_flags or session["idx"] + 1 >= len(scale.items): + # Clinical protocol (crisis_blocking): a crisis flag ends the scale at once. Demo default: the scale is finished + # first and the crisis message comes at the end of the questionnaire (engine defers it), blocking nothing. + stop_now = bool(crisis_flags) and get_settings().crisis_blocking + if stop_now or session["idx"] + 1 >= len(scale.items): return await _finish(ctx, scale, session, answers, result, crisis_flags) updated = {**session, "answers": answers, "idx": session["idx"] + 1} sessions = {**(ctx.data.get("scales") or {}), scale.id: updated} @@ -278,7 +282,7 @@ async def _finish( # Crisis path decided by the scale's own flags (deterministic), not by the LLM. risk_screen = ctx.texts.settings.risk_screen_scale if crisis_open(ctx.data): - crisis = Step(next_state="crisis") + crisis = Step(next_state="crisis") if get_settings().crisis_blocking else Step() else: crisis = open_crisis(ctx, categories=[], source=f"scale:{scale.id}") step = done.then(crisis) diff --git a/app/integrations/notify.py b/app/integrations/notify.py index 29c548f..248970b 100644 --- a/app/integrations/notify.py +++ b/app/integrations/notify.py @@ -92,9 +92,9 @@ class TelegramDoctorChat: def _get_sender(self) -> ChatSender: if self._sender is None: - from aiogram import Bot + from app.channels.telegram import make_bot - self._bot = Bot(self._bot_token) + self._bot = make_bot(self._bot_token) self._sender = self._bot.send_message return self._sender diff --git a/app/llm/agent/__init__.py b/app/llm/agent/__init__.py index 379b252..5f25795 100644 --- a/app/llm/agent/__init__.py +++ b/app/llm/agent/__init__.py @@ -21,6 +21,7 @@ from pydantic import BaseModel from app.config import Settings, get_settings from app.core.types import ScrubbedText +from app.llm.agent import stream from app.llm.agent.prompts import SYSTEM_PROMPT from app.llm.agent.providers import ( GigaChatToolProvider, @@ -127,6 +128,7 @@ async def answer_chat( return None if is_small_talk(question): return AgentAnswer(text=SMALL_TALK_REPLY, sources=[], tools=[], provider="small_talk") + stream.emit("status", "Думаю над вопросом…") ctx = ToolContext(token=token) messages: list[dict[str, Any]] = [ {"role": "system", "content": SYSTEM_PROMPT + await _patient_brief(ctx)}, @@ -159,6 +161,7 @@ async def answer_chat( } ) for c in turn.tool_calls[:MAX_TOOL_CALLS_PER_TURN]: + stream.emit("status", stream.TOOL_STATUS.get(c.name, "Проверяю данные…")) result = await run_tool(ctx, c.name, c.arguments) messages.append({"role": "tool", "tool_call_id": c.id, "name": c.name, "content": result}) log.warning("chat agent: step limit reached tools=%s", ",".join(ctx.used)) diff --git a/app/llm/agent/providers.py b/app/llm/agent/providers.py index 8bd5852..63d4f01 100644 --- a/app/llm/agent/providers.py +++ b/app/llm/agent/providers.py @@ -16,6 +16,7 @@ import httpx from pydantic import BaseModel, Field from app.config import Settings +from app.llm.agent import stream log = logging.getLogger(__name__) @@ -53,6 +54,8 @@ class OpenRouterProvider: self._client = client async def chat(self, messages: list[dict[str, Any]], tools: list[dict[str, Any]]) -> AssistantTurn: + if stream.active(): + return await self._chat_stream(messages, tools) body: dict[str, Any] = {"model": self._model, "messages": messages, "tools": tools, "temperature": 0.1} body["max_tokens"] = 900 if not any(m["role"] == "tool" for m in messages): @@ -174,6 +177,78 @@ class GigaChatToolProvider: return AssistantTurn(content=msg.content or None, tool_calls=calls) +# ------------------------------------------------------------------------------------------------ streaming + + +async def _iter_sse(resp: httpx.Response): + async for line in resp.aiter_lines(): + if not line.startswith("data:"): + continue # comments («: OPENROUTER PROCESSING») and blank lines + data = line[5:].strip() + if data == "[DONE]": + return + try: + yield json.loads(data) + except json.JSONDecodeError: + continue + + +async def _openrouter_stream(self: "OpenRouterProvider", messages: list[dict[str, Any]], tools: list[dict[str, Any]]): + """Same request with `stream: true`: content deltas go to the live channel, tool calls are assembled.""" + body: dict[str, Any] = { + "model": self._model, + "messages": messages, + "tools": tools, + "temperature": 0.1, + "max_tokens": 900, + "stream": True, + } + if not any(m["role"] == "tool" for m in messages): + body["tool_choice"] = "required" + headers = {"Authorization": f"Bearer {self._key}", "X-Title": "PTSRaiAgent prototype (synthetic data)"} + content: list[str] = [] + calls: dict[int, dict[str, Any]] = {} + try: + client = self._client or httpx.AsyncClient(timeout=self._timeout, proxy=self._proxy) + try: + async with client.stream("POST", self._url, json=body, headers=headers, timeout=self._timeout) as resp: + if resp.status_code != 200: + await resp.aread() + raise ProviderError(f"http {resp.status_code}") + async for chunk in _iter_sse(resp): + choice = (chunk.get("choices") or [{}])[0] + delta = choice.get("delta") or {} + if delta.get("content"): + content.append(delta["content"]) + stream.emit("delta", delta["content"]) + for tc in delta.get("tool_calls") or []: + slot = calls.setdefault(tc.get("index", 0), {"id": "", "name": "", "args": ""}) + slot["id"] = tc.get("id") or slot["id"] + fn = tc.get("function") or {} + slot["name"] += fn.get("name") or "" + slot["args"] += fn.get("arguments") or "" + finally: + if self._client is None: + await client.aclose() + except httpx.HTTPError as exc: + raise ProviderError(type(exc).__name__) from exc + out = [] + for i in sorted(calls): + c = calls[i] + try: + args = json.loads(c["args"] or "{}") + except json.JSONDecodeError: + args = {} + out.append(ToolCall(id=c["id"] or c["name"] or f"call{i}", name=c["name"], arguments=args)) + if out and content: + stream.emit("reset", None) # a preface before tool calls is not the answer + log.info("openrouter stream ok calls=%d chars=%d", len(out), sum(map(len, content))) + return AssistantTurn(content="".join(content) or None, tool_calls=out) + + +OpenRouterProvider._chat_stream = _openrouter_stream # type: ignore[attr-defined] + + # ------------------------------------------------------------------------------------------------ offline mock _DRUG_WORDS = re.compile( diff --git a/app/llm/agent/stream.py b/app/llm/agent/stream.py new file mode 100644 index 0000000..bf1bbd2 --- /dev/null +++ b/app/llm/agent/stream.py @@ -0,0 +1,32 @@ +"""Live progress of the chat agent for streaming channels (web SSE). + +The SSE endpoint sets a queue in the context; the agent pushes events while it works: + status — a tool step in plain Russian («Смотрю вашу карту ЕМИАС…») + delta — a chunk of the answer as the model writes it + reset — the streamed text so far was a preface before tool calls: clear it +Channels without a queue (Telegram, tests) are unaffected. Deltas are the raw model text: the final reply +(forbidden-output filter, sources footer) always comes in the closing `done` event and replaces it. +""" + +import asyncio +from contextvars import ContextVar +from typing import Any + +STREAM: ContextVar[asyncio.Queue[tuple[str, Any]] | None] = ContextVar("agent_stream", default=None) + +TOOL_STATUS = { + "get_patient_record": "Смотрю вашу карту ЕМИАС…", + "check_drug_for_patient": "Проверяю препарат по вашей карте: болезни, непереносимости, другие лекарства…", + "get_drug_info": "Открываю инструкцию к препарату (ГРЛС)…", + "search_guidelines": "Ищу в клинических рекомендациях Минздрава…", +} + + +def active() -> bool: + return STREAM.get() is not None + + +def emit(kind: str, payload: Any) -> None: + q = STREAM.get() + if q is not None: + q.put_nowait((kind, payload)) diff --git a/config/instruments/red_flags.yaml b/config/instruments/red_flags.yaml index 66fa0be..d6002f5 100644 --- a/config/instruments/red_flags.yaml +++ b/config/instruments/red_flags.yaml @@ -17,6 +17,8 @@ items: - id: rf_weapons text: Есть ли у Вас дома или рядом доступ к оружию? category: null + # Asked and stored, but NOT a stop rule (team decision 2026-10-04: weapon ownership is common in Russia). + trigger: false - id: rf_violence text: Угрожает ли Вам сейчас кто-то дома или применяет к Вам насилие? category: domestic_violence diff --git a/config/rules.yaml b/config/rules.yaml index a8cce9a..5357834 100644 --- a/config/rules.yaml +++ b/config/rules.yaml @@ -47,6 +47,7 @@ rules: reason_ru: "C-SSRS: высокий риск (намерение/план или поведение)" - id: weapon_access_risk + enabled: false # team decision 2026-10-04: weapon ownership is common in Russia — the answer triggers nothing description: > Access to weapons together with any suicide-risk signal (recent PHQ-9 item 9 >= 1 or C-SSRS >= moderate). Means restriction is one of the most evidence-based suicide-prevention measures → crisis safety block. diff --git a/config/texts/veteran_onboarding.yaml b/config/texts/veteran_onboarding.yaml index 09eeee3..30d033d 100644 --- a/config/texts/veteran_onboarding.yaml +++ b/config/texts/veteran_onboarding.yaml @@ -14,7 +14,8 @@ status: "черновик, утверждает психиатр (#8)" settings: # Флаги шкал (config/scales/*.yaml), которые считаются сигналом риска для блока безопасного хранения оружия. # Положительный п.9 PHQ-9 и C-SSRS — суицидальный риск; AUDIT-C — алкоголь как фактор риска при доступе к оружию. - weapon_risk_flags: [phq9_item9_positive, cssrs_high, cssrs_moderate, auditc_positive] + # The weapon question exists but affects nothing (team decision 2026-10-04): no safety block, no trigger. + weapon_risk_flags: [] intro: text: |- diff --git a/docker-compose.yml b/docker-compose.yml index 2d1d3d4..79bb6ad 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -47,5 +47,14 @@ services: ports: - "8501:8501" + web: + # Patient web app (React) — built inside Docker, no Node needed on the host. + build: ./web + depends_on: + app: + condition: service_healthy + ports: + - "8080:80" + volumes: pgdata: diff --git a/infra/deploy/docker-compose.server.yml b/infra/deploy/docker-compose.server.yml index 24a835f..440bcca 100644 --- a/infra/deploy/docker-compose.server.yml +++ b/infra/deploy/docker-compose.server.yml @@ -28,3 +28,7 @@ services: - "127.0.0.1:8511:8501" restart: unless-stopped mem_limit: 768m + + web: + # On the server the host nginx serves the web bundle itself (deploy.sh builds web/dist). + profiles: ["local-only"] diff --git a/pyproject.toml b/pyproject.toml index e1afa3a..7d952b1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -20,6 +20,7 @@ dependencies = [ "pandas>=2.2", "qrcode[pil]>=7.4", "tzdata>=2026.5", + "aiohttp-socks>=0.8", ] [dependency-groups] diff --git a/tests/conftest.py b/tests/conftest.py index 2bf9991..b3e47c5 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -12,6 +12,8 @@ def _no_live_llm(monkeypatch: pytest.MonkeyPatch): monkeypatch.setenv("CHAT_AGENT", "off") monkeypatch.setenv("OPENROUTER_API_KEY", "") monkeypatch.setenv("LLM_PROVIDER", "mock") + # Tests cover the clinical protocol by default (blocking crisis); demo-mode tests flip it explicitly. + monkeypatch.setenv("CRISIS_BLOCKING", "true") get_settings.cache_clear() yield get_settings.cache_clear() diff --git a/tests/test_agent_stream.py b/tests/test_agent_stream.py new file mode 100644 index 0000000..8258873 --- /dev/null +++ b/tests/test_agent_stream.py @@ -0,0 +1,83 @@ +"""Live agent output: OpenRouter stream parsing and the SSE endpoint (status → delta → done).""" + +import asyncio +import json + +import httpx +import pytest +from httpx import ASGITransport, AsyncClient + +from app.config import Settings +from app.db import session +from app.db.bootstrap import init_db +from app.llm import agent +from app.llm.agent import stream +from app.llm.agent.providers import OpenRouterProvider +from app.main import app + +MEM = "sqlite+aiosqlite:///:memory:" + + +def _sse(*chunks: dict) -> bytes: + lines = [": OPENROUTER PROCESSING", ""] + lines += [f"data: {json.dumps(c, ensure_ascii=False)}" for c in chunks] + ["data: [DONE]", ""] + return "\n\n".join(lines).encode() + + +async def test_openrouter_stream_deltas_and_tool_calls(): + body = _sse( + {"choices": [{"delta": {"content": "Сейчас "}}]}, + { + "choices": [ + { + "delta": { + "tool_calls": [{"index": 0, "id": "c1", "function": {"name": "get_patient_", "arguments": ""}}] + } + } + ] + }, + {"choices": [{"delta": {"tool_calls": [{"index": 0, "function": {"name": "record", "arguments": "{}"}}]}}]}, + ) + client = httpx.AsyncClient(transport=httpx.MockTransport(lambda r: httpx.Response(200, content=body))) + provider = OpenRouterProvider(Settings(_env_file=None, openrouter_api_key="k"), client=client) + q: asyncio.Queue = asyncio.Queue() + token = stream.STREAM.set(q) + try: + turn = await provider.chat([{"role": "user", "content": "x"}], []) + finally: + stream.STREAM.reset(token) + assert [c.name for c in turn.tool_calls] == ["get_patient_record"] + events = [q.get_nowait() for _ in range(q.qsize())] + assert ("delta", "Сейчас ") in events and ("reset", None) in events + + +@pytest.fixture +async def client(): + await session.configure(identity_url=MEM, clinical_url=MEM) + await init_db() + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://t") as c: + yield c + await session.configure() + + +async def test_sse_endpoint_streams_steps_then_final(client, monkeypatch): + async def fake(token, text, **kw): + stream.emit("status", stream.TOOL_STATUS["get_patient_record"]) + stream.emit("delta", "Ответ ") + stream.emit("delta", "агента") + return agent.AgentAnswer(text="Ответ агента", sources=[], tools=["get_patient_record"], provider="fake") + + monkeypatch.setattr(agent, "answer_chat", fake) + token = (await client.post("/api/v1/auth/demo/fresh", json={})).json()["token"] + kinds = [] + async with client.stream( + "POST", f"/api/v1/dialog/{token}/events/stream", json={"type": "text", "text": "Можно ибупрофен?"} + ) as resp: + assert resp.headers["content-type"].startswith("text/event-stream") + async for line in resp.aiter_lines(): + if line.startswith("event:"): + kinds.append(line[6:].strip()) + if line.startswith("data:") and kinds and kinds[-1] == "done": + final = json.loads(line[5:]) + assert kinds[0] == "status" and "delta" in kinds and kinds[-1] == "done" + assert any("Ответ агента" in r["text"] for r in final) diff --git a/tests/test_demo_crisis_deferred.py b/tests/test_demo_crisis_deferred.py new file mode 100644 index 0000000..a8f2587 --- /dev/null +++ b/tests/test_demo_crisis_deferred.py @@ -0,0 +1,36 @@ +"""Demo mode (crisis_blocking off): a crisis signal inside a questionnaire does not stop it; the crisis message +comes at the end and blocks nothing; the weapon question triggers nothing.""" + +import pytest + +from app.db.clinical import Alert +from tests.test_dialog_intake import answer_button, complete_existing, press, released, rows # noqa: F401 + + +@pytest.fixture +def non_blocking(monkeypatch): + from app.config import get_settings + + monkeypatch.setattr(get_settings(), "crisis_blocking", False) + + +async def test_red_flag_does_not_stop_the_questionnaire(released, non_blocking): # noqa: F811 + replies = await press(released, "intake:start") + replies = await press(released, answer_button(replies, {"rf_psychosis": 0})) # «Да» on a red flag + assert not any(r.kind == "crisis" for r in replies), "no crisis window in the middle of the test" + assert any(r.buttons for r in replies), "the next question is asked" + (row,) = await rows() + assert row.status != "interrupted" and row.details["pending_urgent"]["code"] == "red_flag_psychosis" + replies = await complete_existing(released, replies, {}) + assert any(r.kind == "crisis" for r in replies), "the crisis message comes at the end" + assert any(a.code == "red_flag_psychosis" and a.level == "acute" for a in await rows(Alert)) + + +async def test_weapon_answer_triggers_nothing(released, non_blocking): # noqa: F811 + replies = await press(released, "intake:start") + for _ in range(2): # psychosis, harm to others: «Нет» + replies = await press(released, answer_button(replies)) + replies = await press(released, answer_button(replies, {"rf_weapons": 0})) # «Да» on weapons + assert not any(r.kind == "crisis" for r in replies) + (row,) = await rows() + assert not row.details.get("pending_urgent") diff --git a/tests/test_dialog_veteran_onboarding.py b/tests/test_dialog_veteran_onboarding.py index a181158..55c4d4c 100644 --- a/tests/test_dialog_veteran_onboarding.py +++ b/tests/test_dialog_veteran_onboarding.py @@ -268,7 +268,7 @@ async def test_auditc_sex_declined_uses_default_threshold(dbs_nosex: None, engin assert (await results())["auditc"].severity == "positive" # unknown sex → sensitive threshold ≥ 3 # Weapon at home + AUDIT-C positive = risk signal → safe-storage block. r = await press(engine, "vonb:weapon:self") - assert VT.weapon.safety_block_self in [x.text for x in r] + assert VT.weapon.safety_block_self not in [x.text for x in r] # the weapon answer affects nothing # ----------------------------------------------------------------------------------------------- weapon / safety diff --git a/tests/test_rules.py b/tests/test_rules.py index b219b76..f513896 100644 --- a/tests/test_rules.py +++ b/tests/test_rules.py @@ -125,8 +125,9 @@ def test_cssrs_levels() -> None: @pytest.mark.parametrize( ("weapon", "cssrs", "item9", "fires"), [ - (True, "moderate", None, True), - (True, None, 1, True), + # Team decision 2026-10-04: weapon access triggers nothing (rule disabled in config/rules.yaml). + (True, "moderate", None, False), + (True, None, 1, False), (True, "low", 0, False), (False, "high", 2, False), (None, "moderate", None, False), diff --git a/tests/test_services_monitoring.py b/tests/test_services_monitoring.py index 27bd254..123a78a 100644 --- a/tests/test_services_monitoring.py +++ b/tests/test_services_monitoring.py @@ -223,10 +223,9 @@ async def test_weapon_access_from_dialog_facts(db: MemoryNotifier) -> None: assert snap is not None and snap.weapon_access is True await scale("cssrs_screen", 3, severity="moderate", scheduled=False) actions = await monitoring.RuleMonitoring().after_data_change(TOKEN) - assert {a.code for a in actions if a.type == "crisis"} == {"weapon_access_risk"} - plan = next(a for a in actions if a.type == "show_safety_plan") - assert plan.params["block"] == "means_restriction" - assert await color() == "red" + # Team decision 2026-10-04: weapon access is stored but triggers nothing (no crisis, no safety block). + assert "weapon_access_risk" not in {a.code for a in actions} + assert not any(a.type == "show_safety_plan" and a.params.get("block") == "means_restriction" for a in actions) async def test_snapshot_mapping_and_unknown_patient(db: MemoryNotifier) -> None: diff --git a/tests/test_telegram.py b/tests/test_telegram.py index 77b9065..bbcb30f 100644 --- a/tests/test_telegram.py +++ b/tests/test_telegram.py @@ -22,6 +22,7 @@ from app.channels.telegram import ( render_markup, split_text, ) +from app.config import get_settings from app.db import identity_repo, session from app.db.bootstrap import init_db from app.dialog.contracts import Button, Event, Reply @@ -163,7 +164,10 @@ async def test_verified_token_without_identity_record(dbs: None, engine: Recordi assert sender.texts == [load_texts().code_invalid] -async def test_unbound_chat_gets_not_connected(dbs: None, engine: RecordingEngine) -> None: +async def test_unbound_chat_gets_not_connected( + dbs: None, engine: RecordingEngine, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(get_settings(), "demo_clock", False) sender = FakeSender() channel = TelegramChannel(sender) await channel.on_start(CHAT, None) @@ -251,6 +255,18 @@ async def test_outbox_pushes_to_bound_chat_and_falls_back(dbs: None, engine: Rec assert [r.text for r in fallback.drain(TOKEN2)] == ["Как вы сегодня?"] +async def test_start_without_code_on_demo_stand_binds_a_demo_copy( + dbs: None, engine: RecordingEngine, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(get_settings(), "demo_clock", True) + sender = FakeSender() + channel = TelegramChannel(sender) + await channel.on_start(CHAT, None) + token = await channel.directory.token_for(CHAT) + assert token is not None and token.startswith("tok-andrey-") + assert [c[1].type for c in engine.calls] == ["start"] + + async def test_start_is_noop_without_token(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(telegram, "get_settings", lambda: type("S", (), {"tg_bot_token": ""})()) before = integrations.outbox @@ -261,7 +277,9 @@ async def test_start_is_noop_without_token(monkeypatch: pytest.MonkeyPatch) -> N async def test_start_stop_swaps_outbox(monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.setattr(telegram, "get_settings", lambda: type("S", (), {"tg_bot_token": "123:fake"})()) + monkeypatch.setattr( + telegram, "get_settings", lambda: type("S", (), {"tg_bot_token": "123:fake", "tg_proxy_url": ""})() + ) async def fake_polling(self: object, *bots: object, **kwargs: object) -> None: import asyncio diff --git a/tests/test_telegram_proxy.py b/tests/test_telegram_proxy.py new file mode 100644 index 0000000..b09ec48 --- /dev/null +++ b/tests/test_telegram_proxy.py @@ -0,0 +1,16 @@ +"""TG_PROXY_URL: the bot session goes through the proxy when it is set.""" + +from app.channels import telegram +from app.config import get_settings + +TOKEN = "123456:" + "A" * 35 + + +def test_make_bot_without_proxy(monkeypatch): + monkeypatch.setattr(get_settings(), "tg_proxy_url", "") + assert telegram.make_bot(TOKEN).session.proxy is None + + +def test_make_bot_with_proxy(monkeypatch): + monkeypatch.setattr(get_settings(), "tg_proxy_url", "http://user:pass@127.0.0.1:8888") + assert telegram.make_bot(TOKEN).session.proxy == "http://user:pass@127.0.0.1:8888" diff --git a/uv.lock b/uv.lock index fa2c0d6..74d9c09 100644 --- a/uv.lock +++ b/uv.lock @@ -122,6 +122,19 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/b2/0e/50d6e6471cd31edce8b282bdec59375a3a69124d8a989a0b1313355cae52/aiohttp-3.14.3-cp313-cp313-win_arm64.whl", hash = "sha256:ddcac3c6b382e81f1dd0499199d4136b877beb4cb5ef770bbbfba56c4b8f55d2", size = 447676, upload-time = "2026-07-23T01:55:40.451Z" }, ] +[[package]] +name = "aiohttp-socks" +version = "0.12.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "aiohttp" }, + { name = "python-socks" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/18/1d/a306e0111222180e60f17131a3f5d9bc694dd999a8115959a7dd76c2238e/aiohttp_socks-0.12.0.tar.gz", hash = "sha256:3caf9f5a4164611122d412bc11b2f9114fd29c85e1ba27bb38060d3c236bdc8d", size = 12061, upload-time = "2026-08-12T04:43:15.791Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/86/64/ca6289632020523ea1841f01363f83ada10eeb0f21dd931fc1118dd85668/aiohttp_socks-0.12.0-py3-none-any.whl", hash = "sha256:ba6f95ec775c761d87f8578ab48f137d0457c676da104984202bf75e747d5ee6", size = 10693, upload-time = "2026-08-12T04:43:14.524Z" }, +] + [[package]] name = "aiosignal" version = "1.4.0" @@ -1075,6 +1088,7 @@ version = "0.1.0" source = { editable = "." } dependencies = [ { name = "aiogram" }, + { name = "aiohttp-socks" }, { name = "apscheduler" }, { name = "asyncpg" }, { name = "fastapi" }, @@ -1103,6 +1117,7 @@ dev = [ [package.metadata] requires-dist = [ { name = "aiogram", specifier = ">=3.10" }, + { name = "aiohttp-socks", specifier = ">=0.8" }, { name = "apscheduler", specifier = ">=3.10,<4" }, { name = "asyncpg", specifier = ">=0.29" }, { name = "fastapi", specifier = ">=0.115" }, @@ -1340,6 +1355,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/e1/04/e8135ebd1ad02c56ec633277529b2602ff99ff634be76cdba5744cf554fd/python_multipart-0.0.32-py3-none-any.whl", hash = "sha256:ff6d3f776f16878c894e52e107296ffc890e913c611b1a4ec6c44e2821fe2e23", size = 30042, upload-time = "2026-06-04T16:18:57.319Z" }, ] +[[package]] +name = "python-socks" +version = "3.1.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/04/ad/484ffb79532517b11a90af38647c38652224650b31a7ae1cedd5a418d8ab/python_socks-3.1.1.tar.gz", hash = "sha256:8d3e817cdbe858dc0bb8c8fdc8e79b6ce37acce110d33374c6f57a675cc9029e", size = 232781, upload-time = "2026-09-08T13:03:31.058Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/3b/23/2c2cef1b4313c55d1713201acd4ee2043fb2cf22e546ca9d36bf4317faea/python_socks-3.1.1-py3-none-any.whl", hash = "sha256:327e0d6378702c73a7790bf732e9f01392f17b48c7348a50b5bd1f710c2df1be", size = 49566, upload-time = "2026-09-08T13:03:29.604Z" }, +] + [[package]] name = "pyyaml" version = "6.0.3" diff --git a/web/.dockerignore b/web/.dockerignore new file mode 100644 index 0000000..b431156 --- /dev/null +++ b/web/.dockerignore @@ -0,0 +1,3 @@ +node_modules +dist +.vite diff --git a/web/Dockerfile b/web/Dockerfile new file mode 100644 index 0000000..def43b0 --- /dev/null +++ b/web/Dockerfile @@ -0,0 +1,12 @@ +# Patient web app: build with Node in a throwaway stage, serve the static bundle with nginx (+ /api → app). +# The jury only needs Docker: `make up` builds and starts it (http://localhost:8080). +FROM node:20-alpine AS build +WORKDIR /w +COPY package.json package-lock.json ./ +RUN npm ci --no-audit --no-fund +COPY . . +RUN npm run build + +FROM nginx:1.27-alpine +COPY nginx.conf /etc/nginx/conf.d/default.conf +COPY --from=build /w/dist /usr/share/nginx/html diff --git a/web/nginx.conf b/web/nginx.conf new file mode 100644 index 0000000..35d72de --- /dev/null +++ b/web/nginx.conf @@ -0,0 +1,18 @@ +# Same-origin API: the app calls /api/v1/... — proxied to the FastAPI container (SSE streaming unbuffered). +server { + listen 80; + root /usr/share/nginx/html; + index index.html; + + location / { + try_files $uri $uri/ /index.html; + } + + location /api/ { + proxy_pass http://app:8000; + proxy_set_header Host $host; + proxy_http_version 1.1; + proxy_buffering off; + proxy_read_timeout 120s; + } +} diff --git a/web/src/api/client.ts b/web/src/api/client.ts index a0cb402..ea6b5ad 100644 --- a/web/src/api/client.ts +++ b/web/src/api/client.ts @@ -59,3 +59,53 @@ export const api = { outbox: async (token: string) => (await request<{ replies: Reply[] }>(`/dialog/${encodeURIComponent(token)}/outbox`)).replies, }; + +export type StreamEvent = + | { kind: "status"; text: string } + | { kind: "delta"; text: string } + | { kind: "reset" } + | { kind: "done"; replies: Reply[] }; + +/** POST /dialog/{token}/events/stream (Server-Sent Events): agent steps and the answer as it is written, + * then the final replies. Resolves with the final replies; throws if the stream breaks (caller falls back). */ +export async function sendEventStream( + token: string, + event: DialogEvent, + onEvent: (e: StreamEvent) => void, +): Promise { + const res = await fetch(`${BASE}/dialog/${token}/events/stream`, { + method: "POST", + headers: { "Content-Type": "application/json", Accept: "text/event-stream" }, + body: JSON.stringify(event), + }); + if (!res.ok || !res.body) throw new ApiError(res.status, res.statusText); + const reader = res.body.getReader(); + const decoder = new TextDecoder(); + let buf = ""; + for (;;) { + const { value, done } = await reader.read(); + if (done) break; + buf += decoder.decode(value, { stream: true }); + let cut: number; + while ((cut = buf.indexOf("\n\n")) >= 0) { + const block = buf.slice(0, cut); + buf = buf.slice(cut + 2); + let kind = "message"; + let data = ""; + for (const line of block.split("\n")) { + if (line.startsWith("event:")) kind = line.slice(6).trim(); + else if (line.startsWith("data:")) data += line.slice(5).trim(); + } + if (!data) continue; + const payload = JSON.parse(data); + if (kind === "status") onEvent({ kind, text: String(payload) }); + else if (kind === "delta") onEvent({ kind, text: String(payload) }); + else if (kind === "reset") onEvent({ kind: "reset" }); + else if (kind === "done") { + onEvent({ kind: "done", replies: payload as Reply[] }); + return payload as Reply[]; + } else if (kind === "error") throw new ApiError(500, "stream error"); + } + } + throw new ApiError(0, "stream closed"); +} diff --git a/web/src/components/AppShell.tsx b/web/src/components/AppShell.tsx index a93786e..2b89534 100644 --- a/web/src/components/AppShell.tsx +++ b/web/src/components/AppShell.tsx @@ -16,6 +16,7 @@ import { Icon, LogoMark, type IconName } from "./Icon"; const TABS: { to: string; label: string; icon: IconName }[] = [ { to: "/today", label: ru.nav.today, icon: "home" }, + { to: "/messages", label: ru.nav.chat, icon: "message" }, { to: "/meds", label: ru.nav.meds, icon: "pill" }, { to: "/doctor", label: ru.nav.doctor, icon: "doctor" }, { to: "/profile", label: ru.nav.profile, icon: "user" }, diff --git a/web/src/content/ru.ts b/web/src/content/ru.ts index f5288dc..b6f2aff 100644 --- a/web/src/content/ru.ts +++ b/web/src/content/ru.ts @@ -16,6 +16,7 @@ export const ru = { doctor: "Врач", profile: "Профиль", messages: "Сообщения", + chat: "Чат", signIn: "Вход", help: "Помощь", }, @@ -251,6 +252,7 @@ export const ru = { noticeTitle: "Опрос приостановлен", }, messages: { + typing: "Помощник печатает", title: "Сообщения", inputLabel: "Сообщение", inputPlaceholder: "Напишите сообщение", diff --git a/web/src/dialog/DialogContext.tsx b/web/src/dialog/DialogContext.tsx index 13d4793..0503642 100644 --- a/web/src/dialog/DialogContext.tsx +++ b/web/src/dialog/DialogContext.tsx @@ -1,6 +1,6 @@ import { createContext, useCallback, useContext, useEffect, useMemo, useRef, useState, type ReactNode } from "react"; -import { api } from "../api/client"; +import { api, sendEventStream } from "../api/client"; import type { DialogEvent, Reply } from "../api/types"; import { useSession } from "../session/SessionContext"; import { sessionStore, type ChatEntry, type DialogSnapshot } from "../session/store"; @@ -17,8 +17,15 @@ let seq = 0; const newId = () => `${Date.now().toString(36)}-${(seq++).toString(36)}`; const asEntries = (replies: Reply[]): ChatEntry[] => replies.map((reply) => ({ id: newId(), from: "assistant", reply })); +/** The agent at work: its steps and the answer as it is being written (cleared when the final reply lands). */ +export interface LiveReply { + status: string[]; + text: string; +} + interface DialogApi extends DialogSnapshot { busy: boolean; + live: LiveReply | null; error: boolean; started: boolean; /** Sends an event; `echo` is what the patient «said» (button label / text) for the history. */ @@ -36,6 +43,7 @@ export function DialogProvider({ children }: { children: ReactNode }) { const [owner, setOwner] = useState(token); const [snap, setSnap] = useState(() => (token ? sessionStore.loadDialog(token) : EMPTY)); const [busy, setBusy] = useState(false); + const [live, setLive] = useState(null); const [error, setError] = useState(false); const tokenRef = useRef(token); tokenRef.current = token; @@ -58,7 +66,35 @@ export function DialogProvider({ children }: { children: ReactNode }) { setBusy(true); setError(false); try { - const replies = await api.sendEvent(t, event); + let replies: Reply[]; + if (event.type === "text") { + // Free text may go to the agent: stream its steps and the answer so the viewer sees it working. + const said = echo; + if (said) { + // Show the patient's message right away, while the agent is still working. + setSnap((prev) => ({ + ...prev, + entries: [...prev.entries, { id: newId(), from: "patient" as const, text: said }], + })); + echo = undefined; + } + setLive({ status: [], text: "" }); + try { + replies = await sendEventStream(t, event, (e) => + setLive((prev) => { + const cur = prev ?? { status: [], text: "" }; + if (e.kind === "status") return { ...cur, status: [...cur.status, e.text] }; + if (e.kind === "delta") return { ...cur, text: cur.text + e.text }; + if (e.kind === "reset") return { ...cur, text: "" }; + return cur; + }), + ); + } catch { + replies = await api.sendEvent(t, event); // proxies without SSE support: plain request + } + } else { + replies = await api.sendEvent(t, event); + } setSnap((prev) => ({ entries: [ ...prev.entries, @@ -73,6 +109,7 @@ export function DialogProvider({ children }: { children: ReactNode }) { setError(true); return false; } finally { + setLive(null); setBusy(false); } }, @@ -102,8 +139,8 @@ export function DialogProvider({ children }: { children: ReactNode }) { }, [token]); const value = useMemo( - () => ({ ...snap, busy, error, started: snap.entries.length > 0, send, start, markRead }), - [snap, busy, error, send, start, markRead], + () => ({ ...snap, busy, live, error, started: snap.entries.length > 0, send, start, markRead }), + [snap, busy, live, error, send, start, markRead], ); return {children}; } diff --git a/web/src/screens/Messages.tsx b/web/src/screens/Messages.tsx index 3d9f73f..8691f41 100644 --- a/web/src/screens/Messages.tsx +++ b/web/src/screens/Messages.tsx @@ -18,7 +18,7 @@ export function Messages() { useEffect(() => { dialog.markRead(); endRef.current?.scrollIntoView({ block: "end" }); - }, [dialog.entries.length]); + }, [dialog.entries.length, dialog.live?.text, dialog.live?.status.length]); const send = async (e?: FormEvent) => { e?.preventDefault(); @@ -56,6 +56,32 @@ export function Messages() { ), )} + {dialog.live && ( +
  • + + + +
    + {dialog.live.status.map((st, i) => ( +

    + {st} +

    + ))} + {dialog.live.text ? ( +

    + {dialog.live.text} +

    + ) : ( +

    + + + +

    + )} +
    +
    • + )} {quick.length > 0 && (
    diff --git a/web/src/theme/global.css b/web/src/theme/global.css index 672710b..30fce66 100644 --- a/web/src/theme/global.css +++ b/web/src/theme/global.css @@ -103,7 +103,7 @@ button:disabled { inset: auto 0 0 0; z-index: 20; display: grid; - grid-template-columns: repeat(4, 1fr); + grid-template-columns: repeat(5, 1fr); height: calc(var(--nav-h) + var(--safe-bottom)); padding-bottom: var(--safe-bottom); background: var(--c-surface); @@ -2937,3 +2937,19 @@ button.row-card:hover:not(:disabled) { animation: none; } } + + +/* Live agent reply: steps + the answer as it is written (SSE) */ +.bubble--live .live { display: grid; gap: 4px; } +.live__step { margin: 0; font-size: 13px; color: var(--muted, #6b7280); } +.live__step.is-active::after { content: ""; display: inline-block; width: 1em; animation: live-ellipsis 1.2s steps(4) infinite; overflow: hidden; vertical-align: bottom; } +.live__text { margin: 2px 0 0; white-space: pre-wrap; } +.live__cursor { display: inline-block; width: 7px; height: 1em; margin-left: 2px; background: currentColor; vertical-align: text-bottom; animation: live-blink 1s steps(2) infinite; } +.live__dots { display: inline-flex; gap: 4px; margin: 4px 0 0; } +.live__dots span { width: 6px; height: 6px; border-radius: 50%; background: var(--muted, #9ca3af); animation: live-dot 1.2s infinite ease-in-out; } +.live__dots span:nth-child(2) { animation-delay: .15s; } +.live__dots span:nth-child(3) { animation-delay: .3s; } +@keyframes live-blink { 50% { opacity: 0; } } +@keyframes live-dot { 0%, 80%, 100% { opacity: .3; transform: translateY(0); } 40% { opacity: 1; transform: translateY(-3px); } } +@keyframes live-ellipsis { from { width: 0; } to { width: 1em; } } +.tabbar__item span { font-size: 11px; }