PTSRaiAgent/tests/test_dialog_veteran_onboarding.py

"""Veteran start screening (#36): demo persona Andrey end-to-end through `Engine.handle` on the offline mock LLM.

Connect code → consent → age (known) → intro → trauma exposure → PC-PTSD-5 → PCL-5 → PHQ-9 → AUDIT-C → weapon
question → plain-language result → idle. In-memory SQLite (identity / clinical), fake monitoring (lane B stub).
"""

import re
from collections.abc import AsyncIterator
from pathlib import Path
from typing import Any

import pytest
import yaml

from app.core import clock
from app.core.rules import Action
from app.db import identity_repo, repo, session
from app.db.bootstrap import init_db
from app.dialog.contracts import Event, Reply
from app.dialog.engine import Engine
from app.dialog.flows import veteran_onboarding as vo
from app.dialog.texts import load_dialog_texts
from app.integrations.base import VerifiedPerson, integrations
from app.llm.gateway import LLMGateway
from app.llm.mock import MockTasks

TOKEN = "b" * 32
MEM = "sqlite+aiosqlite:///:memory:"
TEXTS = load_dialog_texts()
VT = vo.load_texts()

# Andrey's screening (docs/product/01_demo_persona.md): PC-PTSD-5 = 4, PCL-5 = 46, PHQ-9 = 14 (item 9 = 0),
# AUDIT-C = 3 (below the male threshold), no weapon at home.
PCPTSD5 = [1, 1, 1, 1, 0]
PCL5 = [3] * 6 + [2] * 14
PHQ9 = [2, 2, 2, 2, 2, 2, 1, 1, 0]
AUDITC = [2, 1, 0]


class FakeMonitoring:
    def __init__(self) -> None:
        self.calls = 0

    async def after_data_change(self, token: str) -> list[Action]:
        self.calls += 1
        return []

    async def is_in_crisis(self, token: str) -> bool:
        return True  # only doctor_ack ends a crisis in these tests


async def _make_patient(sex: str | None) -> None:
    async with session.get_identity_session() as s:
        await identity_repo.create_patient(
            s,
            token=TOKEN,
            display_name="Андрей (синтетика)",
            snils="000-000-000 00",
            clinic_id="c1",
            doctor_id="doc-test",
            now=clock.now(),
            birth_year=1988,
            sex=sex or "m",
        )
    async with session.get_clinical_session() as s:
        await repo.upsert_profile(s, token=TOKEN, doctor_id="doc-test", modules=["veteran"], sex=sex)


@pytest.fixture
async def dbs() -> AsyncIterator[None]:
    await session.configure(identity_url=MEM, clinical_url=MEM)
    await init_db()
    await _make_patient("m")
    yield
    await session.configure()


@pytest.fixture
async def dbs_nosex() -> AsyncIterator[None]:
    await session.configure(identity_url=MEM, clinical_url=MEM)
    await init_db()
    await _make_patient(None)
    yield
    await session.configure()


@pytest.fixture
def mon() -> FakeMonitoring:
    return FakeMonitoring()


@pytest.fixture
def engine(mon: FakeMonitoring) -> Engine:
    return Engine(gateway=LLMGateway(MockTasks()), monitoring=mon)


# ----------------------------------------------------------------------------------------------- helpers


async def press(e: Engine, data: str) -> list[Reply]:
    return await e.handle(TOKEN, Event(type="button", button=data))


async def text(e: Engine, t: str) -> list[Reply]:
    return await e.handle(TOKEN, Event(type="text", text=t))


def buttons(reply: Reply) -> list[str]:
    return [b.data for row in reply.buttons for b in row]


def scale_buttons(replies: list[Reply], scale: str) -> list[str]:
    for reply in reversed(replies):
        found = [d for d in buttons(reply) if d.startswith(f"scale:{scale}:")]
        if found:
            return found
    raise AssertionError(f"no question of {scale} in replies: {[r.text[:40] for r in replies]}")


async def answer(e: Engine, replies: list[Reply], scale: str, values: list[int]) -> list[Reply]:
    """Answer `values` item by item with the buttons of the currently shown question."""
    for value in values:
        choice = [d for d in scale_buttons(replies, scale) if d.endswith(f":{value}")]
        assert choice, f"{scale}: no option {value}"
        replies = await press(e, choice[0])
    return replies


async def get_state() -> tuple[str, dict[str, Any]]:
    async with session.get_clinical_session() as s:
        row = await repo.get_dialog_state(s, TOKEN)
        assert row is not None
        return row.state, dict(row.data)


async def results() -> dict[str, Any]:
    async with session.get_clinical_session() as s:
        return {r.scale: r for r in await repo.list_scale_results(s, TOKEN)}


async def profile() -> Any:
    async with session.get_clinical_session() as s:
        return await repo.get_profile(s, TOKEN)


async def to_intro(e: Engine) -> list[Reply]:
    await e.handle(TOKEN, Event(type="start"))
    return await press(e, "consent:yes")


async def to_weapon(e: Engine, phq9: list[int] = PHQ9) -> list[Reply]:
    await to_intro(e)
    r = await press(e, "vonb:go:yes")
    assert buttons(r[-1]) == ["vonb:exposure:yes", "vonb:exposure:no", "vonb:exposure:declined"]
    r = await press(e, "vonb:exposure:yes")
    r = await answer(e, r, "pcptsd5", PCPTSD5)
    r = await answer(e, r, "pcl5", PCL5)
    r = await answer(e, r, "phq9", phq9)
    return r


# ----------------------------------------------------------------------------------------------- happy path


async def test_andrey_full_onboarding(
    dbs: None, engine: Engine, mon: FakeMonitoring, monkeypatch: pytest.MonkeyPatch
) -> None:
    class Ident:
        async def verify(self, auth_code: str) -> VerifiedPerson | None:
            return VerifiedPerson(token=TOKEN, doctor_id="doc-test") if auth_code == "QR-ANDREY" else None

    monkeypatch.setattr(integrations, "identity", Ident())
    r = await engine.handle(TOKEN, Event(type="start", params={"code": "QR-ANDREY"}))
    assert buttons(r[0]) == ["consent:yes", "consent:no"]

    r = await press(engine, "consent:yes")  # birth year known → adult → veteran onboarding
    assert (await get_state())[0] == vo.START
    assert r[-1].text == VT.intro.text
    assert buttons(r[-1]) == ["vonb:go:yes", "vonb:faq:who", "vonb:faq:unit", "vonb:faq:licences"]

    r = await press(engine, "vonb:go:yes")
    assert (await get_state())[0] == vo.EXPOSURE
    r = await press(engine, "vonb:exposure:yes")
    assert (await get_state())[0] == "scale.pcptsd5"
    r = await answer(engine, r, "pcptsd5", PCPTSD5)
    assert (await get_state())[0] == "scale.pcl5"
    r = await answer(engine, r, "pcl5", PCL5)
    assert (await get_state())[0] == "scale.phq9"
    r = await answer(engine, r, "phq9", PHQ9)
    assert (await get_state())[0] == "scale.auditc"  # sex known from profile → no question
    r = await answer(engine, r, "auditc", AUDITC)
    assert (await get_state())[0] == vo.WEAPON
    weapon_choices = ["none", "self", "others_or_safe", "declined"]
    assert buttons(r[-1]) == [f"vonb:weapon:{c}" for c in weapon_choices]

    r = await press(engine, "vonb:weapon:none")
    state, data = await get_state()
    assert state == "idle"
    texts = [x.text for x in r]
    assert VT.weapon.ack in texts
    summary = next(x for x in r if x.text.startswith(VT.result.intro))
    assert VT.result.ptsd in summary.text and VT.result.mood in summary.text
    assert VT.result.alcohol not in summary.text  # AUDIT-C 3 < 4 for men
    assert not re.search(r"\d", summary.text)  # no scores shown to the patient
    assert r[-1].text == TEXTS.idle.welcome and "menu:checkin" in buttons(r[-1])

    res = await results()
    assert {k: v.total for k, v in res.items()} == {"pcptsd5": 4, "pcl5": 46, "phq9": 14, "auditc": 3}
    assert res["phq9"].item9 == 0 and res["auditc"].severity == "negative"
    assert all(v.complete and v.scheduled for v in res.values())
    assert (await profile()).modules == ["veteran", "depression"]

    facts = data["facts"]
    assert facts["trauma_exposure"] is True
    assert facts["weapon_access"] is False and facts["weapon_access_choice"] == "none"
    assert facts["weapon_safety_block"] is False and "weapon_access_at" in facts
    assert data["veteran_onboarding"].get("stage") is None and "done_at" in data["veteran_onboarding"]
    assert "crisis" not in data
    assert mon.calls >= 5  # every scale + weapon + result


async def test_exposure_no_skips_pcptsd5(dbs: None, engine: Engine) -> None:
    await to_intro(engine)
    await press(engine, "vonb:go:yes")
    r = await press(engine, "vonb:exposure:no")
    assert (await get_state())[0] == "scale.pcl5"  # veterans get PCL-5 regardless
    assert VT.exposure.ack_other in [x.text for x in r]
    r = await answer(engine, r, "pcl5", [0] * 20)
    r = await answer(engine, r, "phq9", [0] * 9)
    r = await answer(engine, r, "auditc", [0, 0, 0])
    r = await press(engine, "vonb:weapon:declined")
    state, data = await get_state()
    assert state == "idle"
    assert "pcptsd5" not in await results()
    assert data["facts"]["trauma_exposure"] is False
    assert data["facts"]["weapon_access"] is None and data["facts"]["weapon_access_choice"] == "declined"
    assert any(VT.result.calm in x.text for x in r)
    assert (await profile()).modules == ["veteran"]  # PHQ-9 0 < 10 → no depression module


async def test_exposure_declined_and_free_text_reasks(dbs: None, engine: Engine) -> None:
    await to_intro(engine)
    await press(engine, "vonb:go:yes")
    r = await text(engine, "ну такое")
    assert (await get_state())[0] == vo.EXPOSURE
    assert VT.buttons_hint in r[-1].text and "vonb:exposure:yes" in buttons(r[-1])
    await press(engine, "vonb:exposure:declined")
    state, data = await get_state()
    assert state == "scale.pcl5" and data["facts"]["trauma_exposure"] is None


async def test_auditc_asks_sex_when_unknown(dbs_nosex: None, engine: Engine) -> None:
    await to_weapon(engine)
    assert (await get_state())[0] == vo.AUDITC
    r = await press(engine, "vonb:sex:f")
    assert (await get_state())[0] == "scale.auditc"
    assert (await profile()).sex == "f"
    await answer(engine, r, "auditc", AUDITC)
    res = await results()
    assert res["auditc"].severity == "positive"  # 3 is positive for women


async def test_auditc_sex_declined_uses_default_threshold(dbs_nosex: None, engine: Engine) -> None:
    await to_weapon(engine)
    r = await press(engine, "vonb:sex:declined")
    assert (await profile()).sex is None
    r = await answer(engine, r, "auditc", AUDITC)
    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 not in [x.text for x in r]  # the weapon answer affects nothing


# ----------------------------------------------------------------------------------------------- weapon / safety


async def test_weapon_without_risk_no_block(dbs: None, engine: Engine) -> None:
    r = await to_weapon(engine)
    r = await answer(engine, r, "auditc", AUDITC)
    r = await press(engine, "vonb:weapon:others_or_safe")
    texts = [x.text for x in r]
    assert VT.weapon.safety_block_stored not in texts and VT.weapon.safety_block_self not in texts
    _, data = await get_state()
    assert data["facts"]["weapon_access"] is True and data["facts"]["weapon_access_choice"] == "others_or_safe"
    assert data["facts"]["weapon_safety_block"] is False


async def test_item9_positive_crisis_then_resume_and_safety_block(dbs: None, engine: Engine) -> None:
    r = await to_weapon(engine, phq9=[2, 2, 2, 2, 2, 2, 1, 1, 1])
    state, data = await get_state()
    # Engine crisis path: fixed script, C-SSRS queued and started while in crisis.
    assert data["crisis"]["open"] is True and data["crisis"]["source"] == "scale:phq9"
    assert any(x.kind == "crisis" for x in r)
    assert state == "scale.cssrs_screen"
    r = await answer(engine, r, "cssrs_screen", [0] * 6)
    assert (await get_state())[0] == "crisis"

    # Doctor acknowledged → idle → the onboarding resumes where it stopped (PHQ-9 already stored → AUDIT-C).
    r = await engine.handle(TOKEN, Event(type="system", system="doctor_ack"))
    assert VT.intro.resume in [x.text for x in r]
    assert (await get_state())[0] == "scale.auditc"
    r = await answer(engine, r, "auditc", AUDITC)
    r = await press(engine, "vonb:weapon:self")
    state, data = await get_state()
    assert VT.weapon.safety_block_self in [x.text for x in r]
    assert data["facts"]["weapon_access"] is True and data["facts"]["weapon_safety_block"] is True
    assert state == "idle"
    assert (await results())["phq9"].item9 == 1


# ----------------------------------------------------------------------------------------------- FAQ


async def test_faq_in_onboarding_and_idle(dbs: None, engine: Engine) -> None:
    await to_intro(engine)
    for key in vo.FAQ_KEYS:
        r = await press(engine, f"vonb:faq:{key}")
        assert r[0].text == getattr(VT.faq, key).text
        assert "vonb:go:yes" in buttons(r[-1])  # the intro is shown again
        assert (await get_state())[0] == vo.START
    await press(engine, "vonb:go:yes")
    r = await press(engine, "vonb:faq:unit")
    assert r[0].text == VT.faq.unit.text and "vonb:exposure:yes" in buttons(r[-1])

    # FAQ buttons keep working after onboarding (idle triggers).
    r = await answer(engine, await press(engine, "vonb:exposure:no"), "pcl5", [0] * 20)
    r = await answer(engine, r, "phq9", [0] * 9)
    r = await answer(engine, r, "auditc", [0, 0, 0])
    await press(engine, "vonb:weapon:none")
    assert (await get_state())[0] == "idle"
    r = await press(engine, "vonb:faq:licences")
    assert r[0].text == VT.faq.licences.text and "menu:checkin" in buttons(r[0])


def test_faq_texts_honest() -> None:
    assert "прямо предусмотренных законом" in VT.faq.unit.text
    assert "лечащий врач" in VT.faq.who.text
    raw = Path(vo.texts_path()).read_text(encoding="utf-8")
    assert raw.count("формулировку проверяет юрист") >= 3
    assert VT.status == "черновик, утверждает психиатр (#8)"


# ----------------------------------------------------------------------------------------------- texts scan

FORBIDDEN = ["диагноз", "расстройств", "психолог", "психиатр", "болезн", "псих"]
FORBIDDEN_SERVICE = ["учёт", "учет", "контрол", "надзор"]


def _strings(node: Any, path: str = "") -> list[tuple[str, str]]:
    if isinstance(node, dict):
        return [s for k, v in node.items() for s in _strings(v, f"{path}.{k}")]
    if isinstance(node, list):
        return [s for v in node for s in _strings(v, path)]
    return [(path, node)] if isinstance(node, str) else []


def test_yaml_forbidden_words() -> None:
    data = yaml.safe_load(Path(vo.texts_path()).read_text(encoding="utf-8"))
    strings = [(p, s) for p, s in _strings(data) if p != ".status"]
    assert strings
    for path, value in strings:
        low = value.lower()
        for word in FORBIDDEN:
            assert word not in low, f"{path}: «{word}»"
        # Quoting the patient's own fear in a FAQ button label is allowed; statements about the service are not.
        if not (path.startswith(".faq.") and path.endswith(".button")):
            for word in FORBIDDEN_SERVICE:
                assert word not in low, f"{path}: «{word}»"