feat(backend): demo scheduler on the virtual clock — doses, missed, check-ins, scales, silence

- app/scheduler/demo.py: run_until(now) replays due and fresh events in time order: dose rows +
  med_reminder, missed after 2 h without an answer → monitoring, daily_checkin, scale_due by the
  config/modules.yaml schedules, daily monitoring pass (silence, missed doses) → reach_out/scale_due
- idempotent via scheduler_events (unique token+key); stale events are not sent; no contact without
  consent or during an open crisis
- config/scheduler.yaml: times and windows; runs while the questionnaire release is off
- POST /demo/clock/advance replays the scheduler up to the new time

Refs #13

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

app/api/routes.py

@@ -25,6 +25,7 @@ from app.api.schemas import (
 from app.core import clock
 from app.dialog.contracts import get_engine
 from app.integrations.base import MemoryOutbox, integrations
+from app.scheduler import jobs as scheduler
 
 router = APIRouter(prefix="/api/v1")
 router.include_router(questionnaire_router)
@@ -99,8 +100,9 @@ async def get_clock() -> ClockOut:
 
 @router.post("/demo/clock/advance", response_model=ClockOut, tags=["demo"])
 async def advance_clock(body: ClockAdvanceIn) -> ClockOut:
-    """Moves the virtual clock; #13 makes the scheduler replay all events up to the new time."""
+    """Moves the virtual clock and replays the scheduler up to the new time (#13)."""
     clock.advance(days=body.days, hours=body.hours)
+    await scheduler.run_until_now()
     return _clock_out()
 
 

app/db/bootstrap.py

@@ -5,6 +5,7 @@ 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 scheduler as _scheduler  # noqa: F401
 from app.db.clinical import ClinicalBase
 from app.db.identity import IdentityBase
 from app.db.session import clinical_engine, identity_engine

app/db/scheduler.py

@@ -0,0 +1,24 @@
+"""Delivery log of the demo scheduler (#13): one row per delivered event, the unique key makes runs idempotent.
+
+Clinical DB, keyed by token only (guardrail #5). Created by `init_db` (`app.db.bootstrap`).
+"""
+
+from datetime import datetime
+
+from sqlalchemy import String, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db.clinical import TOKEN_LEN, BigId, ClinicalBase
+from app.db.types import TZDateTime
+
+
+class SchedulerEvent(ClinicalBase):
+    __tablename__ = "scheduler_events"
+    __table_args__ = (UniqueConstraint("token", "key", name="uq_scheduler_events_token_key"),)
+
+    id: Mapped[int] = mapped_column(BigId, primary_key=True, autoincrement=True)
+    token: Mapped[str] = mapped_column(String(TOKEN_LEN))
+    key: Mapped[str] = mapped_column(String(96))  # e.g. "dose:3:2026-10-06T08:00", "checkin:2026-10-06"
+    kind: Mapped[str] = mapped_column(String(32))  # med_reminder | missed | daily_checkin | scale_due | evaluation
+    due_at: Mapped[datetime] = mapped_column(TZDateTime)
+    done_at: Mapped[datetime] = mapped_column(TZDateTime)

app/dialog/actions.py

@@ -6,7 +6,7 @@ are ignored here. While a crisis is open only crisis-relevant actions (crisis, s
 the patient: no medication tips or visit offers in the middle of a crisis.
 """
 
-from collections.abc import Iterable
+from collections.abc import Awaitable, Callable, Iterable
 from typing import Any
 
 from app.core.rules import Action
@@ -20,6 +20,16 @@ from app.dialog.scale_session import active_scale, add_pending
 _MED_INFO_INTENTS = {"delayed_effect": "quit_intent", "alcohol_interaction": "alcohol_info", "nausea": "nausea_info"}
 _CRISIS_SAFE = {"crisis", "show_safety_plan", "schedule_scale"}
 
+# Optional `med_info` answerer registered by a flow (`flows/meds.py`, #37): answers from the drug instruction
+# (med_qa). Returning None falls back to the phrase template below.
+MedInfoHandler = Callable[[Ctx, Action], Awaitable[Reply | None]]
+_med_info_handler: MedInfoHandler | None = None
+
+
+def register_med_info_handler(handler: MedInfoHandler | None) -> None:
+    global _med_info_handler
+    _med_info_handler = handler
+
 
 def _plan_line(step: Any) -> str:
     if isinstance(step, dict):
@@ -57,9 +67,11 @@ async def execute(ctx: Ctx, actions: Iterable[Action]) -> Step:
             pending = add_pending(data, scale_id, bool(action.params.get("scheduled", False)))
             step = step.then(Step(data={"pending_scales": pending}))
         elif action.type == "med_info":
-            intent = _MED_INFO_INTENTS.get(str(action.params.get("topic")), "default")
+            reply = await _med_info_handler(ctx, action) if _med_info_handler is not None else None
-            text = await ctx.gateway.phrase(intent, {})
+            if reply is None:
-            step = step.then(Step(replies=[Reply(text=text, kind="info")]))
+                intent = _MED_INFO_INTENTS.get(str(action.params.get("topic")), "default")
+                reply = Reply(text=await ctx.gateway.phrase(intent, {}), kind="info")
+            step = step.then(Step(replies=[reply]))
         elif action.type == "suggest_visit":
             text = await ctx.gateway.phrase("book_visit_offer", {})
             button = Button(label=ctx.texts.visit.button, data="menu:visit")

app/dialog/engine.py

@@ -2,6 +2,10 @@
 
 Pipeline for one incoming event (SPEC §3, guardrails #1, #2, #3):
 
+  0. abandoned step     the FSM sits in a question step (check-in, scale) the patient left more than
+                        `settings.abandon_after_hours` ago → back to idle first (stored answers stay; never in
+                        crisis, never in an onboarding chain), then the event is handled as usual.
+
   text event
     1. scrub()            raw text → ScrubbedText. Raw text is never stored, logged or sent anywhere.
     2. consent gate       no valid consent → no LLM, nothing stored; only the local crisis lexicon runs
@@ -25,7 +29,7 @@ Logs carry only the token, states and event codes (guardrail #8).
 import asyncio
 import logging
 from dataclasses import dataclass, field
-from datetime import datetime
+from datetime import datetime, timedelta
 from typing import Any
 from weakref import WeakValueDictionary
 
@@ -45,7 +49,16 @@ from app.dialog.contracts import Event, Reply, register_engine
 from app.dialog.crisis import crisis_open, no_consent_crisis_text, open_crisis
 from app.dialog.flows import load_flows
 from app.dialog.flows.core import close_crisis, revoke
-from app.dialog.fsm import NO_CONSENT_STATES, RESTING_STATES, Ctx, Step, apply_data, enter_hook_for, handler_for
+from app.dialog.fsm import (
+    NO_CONSENT_STATES,
+    RESTING_STATES,
+    Ctx,
+    Step,
+    apply_data,
+    enter_hook_for,
+    handler_for,
+    is_abandonable,
+)
 from app.dialog.texts import DialogTexts, load_dialog_texts
 from app.llm.gateway import LLMGateway, get_gateway
 from app.services import questionnaire as intake_service
@@ -58,6 +71,8 @@ MAX_ENTER_CHAIN = 8
 _patient_locks = WeakValueDictionary()
 # DB unreachable: answer safely instead of failing the channel (connectivity errors only — bugs must surface).
 _DB_DOWN = (OSError, OperationalError, InterfaceError)
+# dialog_state.data key: when the patient last acted in (or the FSM entered) the current step — abandon timeout.
+STEP_AT = "step_at"
 
 
 @dataclass
@@ -109,6 +124,8 @@ class Engine:
         state_before = ctx.state
         run = _Run()
 
+        if self._abandoned(ctx):
+            await self._apply(ctx, self._abandon(ctx), run)
         await self._apply(ctx, await self._route(ctx, event), run)
 
         if run.notify:
@@ -126,6 +143,9 @@ class Engine:
                 await repo.add_message(
                     session, token=token, direction="out", text=scrub(reply.text), ts=ctx.now, fsm_state=ctx.state
                 )
+        if event.type != "system" or ctx.state != state_before:
+            # System events the step ignores (reminders inside a scale) do not count as the patient being there.
+            ctx.data[STEP_AT] = ctx.now.isoformat()
         await self._save_state(ctx)
         task_id = event.params.get("questionnaire_task")
         if event.type == "system" and task_id and run.replies:
@@ -145,6 +165,24 @@ class Engine:
         )
         return run.replies
 
+    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):
+            return False
+        hours = ctx.texts.settings.abandon_after_hours
+        last = ctx.data.get(STEP_AT)
+        if not hours or not last or ctx.now - datetime.fromisoformat(last) < timedelta(hours=hours):
+            return False
+        return is_abandonable(ctx)
+
+    def _abandon(self, ctx: Ctx) -> Step:
+        """Drop the unfinished step's progress (answers already stored in item_responses / daily_checkins stay;
+        an unfinished scale gets no scale_results row) and rest in idle so queued scales may start."""
+        log.info("dialog step abandoned token=%s state=%s", ctx.token, ctx.state)
+        scale_id = scale_session.active_scale(ctx.state)
+        scales = {k: v for k, v in (ctx.data.get("scales") or {}).items() if k != scale_id}
+        return Step(next_state="idle", data={"checkin": None, "scales": scales or None})
+
     async def _route(self, ctx: Ctx, event: Event) -> Step:
         consented = _consented(ctx)
         if event.type == "text":
@@ -307,10 +345,19 @@ class Engine:
         if not pending or ctx.state not in RESTING_STATES:
             return Step()
         screen = ctx.texts.settings.risk_screen_scale
+        dropped: dict[str, Any] = {}
+        if any(p["scale"] == screen for p in pending) and await scale_session.recent_risk_screen(ctx, screen):
+            # Item 9 / rule phq9_item9 / scale_due asked for a C-SSRS the patient completed within the cooldown:
+            # do not repeat the same suicide screen (#56). Crisis script and alerts are unaffected.
+            log.info("dialog risk screen skipped (cooldown) token=%s", ctx.token)
+            pending = [p for p in pending if p["scale"] != screen]
+            dropped = {"pending_scales": pending or None}
+            if not pending:
+                return Step(data=dropped)
         if ctx.state == "crisis":
             pending_screen = [p for p in pending if p["scale"] == screen]
             if not pending_screen:
-                return Step()
+                return Step(data=dropped)
             item = pending_screen[0]
         else:
             item = pending[0]

app/dialog/flows/meds.py

@@ -0,0 +1,465 @@
+"""Medication flow (#37, case scenario «управление приёмом лекарств»). Module-agnostic: every patient with a
+prescription gets it, whatever the clinical module.
+
+    system:med_reminder {dose_id, drug, dose}
+        ──▶ «Принял»                → dose taken, check-in med_taken = true
+        ──▶ «Пропущу»               → dose skipped, med_taken = false ──▶ reason:
+                                       забыл / побочки (→ side effects) / решил бросить (quit_intent) / другое
+        ──▶ «Есть побочные эффекты» → code buttons (тошнота, бессонница, …) or «Другое» → free text
+                                       → check-in side_effects (+ dose.side_effects) → «А препарат сегодня приняли?»
+    menu «Вопрос о лекарстве" ──▶ meds.question (free text) ──▶ med_qa (instruction excerpt) or «уточните у врача»
+    rule action med_info (actions.py hook) ──▶ instruction excerpt for the rule topic + fixed notice
+
+Everything the patient answers is written to the clinical DB (`doses`, `daily_checkins`) and the step asks the
+engine to run monitoring (`notify_monitoring`): the rules (lane B, ADR-007) decide alerts and further actions
+(quit intent → delayed-onset explanation, missed doses → caring contact). Nothing here decides risk.
+
+Reminder buttons carry the dose they refer to (#56): "med:taken:<ref>", "med:skip:<ref>",
+"med:side_effects:<ref>" and the follow-up buttons likewise ("med:se:<code>:<ref>", "med:skip_reason:<r>:<ref>",
+"med:dose:yes:<ref>"), matched by prefix triggers. `<ref>` is the `doses.id` of the reminder or, when there is
+no dose row (prescription only known to the МИС), the reminder's ISO date. A button of an earlier day's dose is
+stale (friendly note, nothing recorded); so are the old payload-less buttons ("med:taken") of earlier releases.
+Check-in answers without a button (daily.med step, free text) go to today's planned dose, if any.
+"""
+
+from datetime import date, timedelta
+from functools import lru_cache
+from pathlib import Path
+from typing import Any
+
+import yaml
+from pydantic import BaseModel, ConfigDict
+from sqlalchemy import select
+
+from app.config import get_settings
+from app.core.privacy import scrub
+from app.core.rules import Action
+from app.db import repo
+from app.db.clinical import DailyCheckin, Dose
+from app.db.clinical import Prescription as PrescriptionRow
+from app.dialog.actions import register_med_info_handler
+from app.dialog.contracts import Button, Event, Reply
+from app.dialog.flows.core import menu_reply
+from app.dialog.fsm import Ctx, Step, abandonable, get_trigger, handler_for, state, trigger, trigger_payload
+from app.integrations.base import integrations
+from app.llm.med_qa import fallback_text, med_qa
+
+QUESTION_STATE = "meds.question"
+SIDE_EFFECT_TEXT_STATE = "meds.side_effect_text"
+SIDE_EFFECT_CODES = ("nausea", "insomnia", "headache", "dizziness", "agitation", "other")
+SKIP_REASONS = ("forgot", "side_effects", "quit", "other")
+
+
+# ------------------------------------------------------------------------------------------- texts
+
+
+class _Strict(BaseModel):
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+
+class ReminderTexts(_Strict):
+    text: str
+    text_no_drug: str
+    taken: str
+    skip: str
+    side_effects: str
+
+
+class SkipTexts(_Strict):
+    question: str
+    reasons: dict[str, str]
+    ack: dict[str, str]
+
+
+class SideEffectTexts(_Strict):
+    question: str
+    options: dict[str, str]
+    questions: dict[str, str]
+    describe: str
+    ack: str
+    repeat: str
+    ask_dose: str
+    dose_yes: str
+    dose_no: str
+
+
+class CheckinMedTexts(_Strict):
+    question: str
+    question_no_drug: str
+    yes: str
+    no: str
+    later: str
+
+
+class QuestionTexts(_Strict):
+    prompt: str
+    no_prescription: str
+
+
+class MedInfoTopic(_Strict):
+    section: str  # med_qa section code (side_effects | onset | alcohol | ...)
+    question: str
+    lead: str = ""
+    fallback: str
+    notice: str = ""
+
+
+class MedsTexts(_Strict):
+    status: str
+    reminder: ReminderTexts
+    taken_ack: str
+    stale: str
+    skip: SkipTexts
+    side_effects: SideEffectTexts
+    checkin: CheckinMedTexts
+    question: QuestionTexts
+    med_info: dict[str, MedInfoTopic]
+
+
+def meds_texts_path() -> Path:
+    return get_settings().config_dir / "texts" / "meds.yaml"
+
+
+@lru_cache
+def load_meds_texts(path: Path | None = None) -> MedsTexts:
+    path = path or meds_texts_path()
+    return MedsTexts.model_validate(yaml.safe_load(path.read_text(encoding="utf-8")))
+
+
+# ------------------------------------------------------------------------------------------- data helpers
+
+
+class Rx(BaseModel):
+    """Active prescription as the dialog needs it (drug INN + dose exactly as the doctor wrote them)."""
+
+    id: int | None
+    drug: str
+    dose: str
+
+
+async def current_prescription(ctx: Ctx) -> Rx | None:
+    """Latest prescription active today: clinical DB first, then the МИС integration (demo: in-memory)."""
+    today = ctx.now.date()
+    rows = await ctx.session.scalars(
+        select(PrescriptionRow).where(PrescriptionRow.token == ctx.token).order_by(PrescriptionRow.id.desc())
+    )
+    for row in rows:
+        if (row.start_date is None or row.start_date <= today) and (row.end_date is None or row.end_date >= today):
+            return Rx(id=row.id, drug=row.drug, dose=row.dose)
+    for p in await integrations.prescriptions.get_prescriptions(ctx.token):
+        if (p.start_date is None or p.start_date <= today) and (p.end_date is None or p.end_date >= today):
+            return Rx(id=None, drug=p.drug, dose=p.dose)
+    return None
+
+
+async def checkin_of(ctx: Ctx, day: date) -> DailyCheckin | None:
+    return await ctx.session.scalar(
+        select(DailyCheckin).where(DailyCheckin.token == ctx.token, DailyCheckin.day == day)
+    )
+
+
+async def save_checkin(ctx: Ctx, **fields: Any) -> DailyCheckin:
+    return await repo.upsert_checkin(ctx.session, token=ctx.token, day=ctx.now.date(), now=ctx.now, **fields)
+
+
+def dose_ref(dose_id: int | None, day: date) -> str:
+    """Button payload naming a dose: its row id, or the reminder day when there is no row."""
+    return str(dose_id) if dose_id is not None else day.isoformat()
+
+
+async def todays_dose(ctx: Ctx) -> Dose | None:
+    """Latest dose row planned for today (local day of `ctx.now`) — target of answers given without a button."""
+    rows = await ctx.session.scalars(select(Dose).where(Dose.token == ctx.token).order_by(Dose.id.desc()).limit(20))
+    today = ctx.now.date()
+    return next((r for r in rows if r.planned_at.astimezone(ctx.now.tzinfo).date() == today), None)
+
+
+async def resolve_ref(ctx: Ctx, ref: str | None) -> tuple[bool, Dose | None]:
+    """(fresh, dose row) for a button payload. Not fresh: missing/foreign/unknown ref or a dose of another day."""
+    if not ref:
+        return False, None
+    if ref.isdigit():
+        row = await ctx.session.get(Dose, int(ref))
+        if row is None or row.token != ctx.token:
+            return False, None
+        return row.planned_at.astimezone(ctx.now.tzinfo).date() == ctx.now.date(), row
+    try:
+        return date.fromisoformat(ref) == ctx.now.date(), None
+    except ValueError:
+        return False, None
+
+
+async def _dose_row(ctx: Ctx, ref: str | None) -> Dose | None:
+    if ref is None:
+        return await todays_dose(ctx)
+    return (await resolve_ref(ctx, ref))[1]
+
+
+async def record_dose(ctx: Ctx, taken: bool, *, status: str | None = None, ref: str | None = None) -> None:
+    """Dose answer → `doses` row (`ref` from the button, else today's) + `daily_checkins.med_taken` (rules read
+    both)."""
+    row = await _dose_row(ctx, ref)
+    if row is not None:
+        row.status = status or ("taken" if taken else "skipped")
+        row.answered_at = ctx.now
+    await save_checkin(ctx, med_taken=taken)
+
+
+async def record_side_effects(ctx: Ctx, codes: list[str], *, ref: str | None = None) -> None:
+    """Merge side-effect codes into today's check-in (rule `side_effect_3d` counts days with any code)."""
+    today = await checkin_of(ctx, ctx.now.date())
+    merged = list(dict.fromkeys([*(today.side_effects if today and today.side_effects else []), *codes]))
+    await save_checkin(ctx, side_effects=merged)
+    row = await _dose_row(ctx, ref)
+    if row is not None:
+        row.side_effects = ",".join(c for c in dict.fromkeys([*(row.side_effects or "").split(","), *codes]) if c)
+
+
+async def side_effect_reply(ctx: Ctx, codes: list[str]) -> Reply:
+    """Reaction to newly reported side effects.
+
+    Same effect as yesterday → short support (the instruction was already shown; persistence over 3 days is the
+    rules' business). New effect with a known instruction question → the excerpt of the drug instruction
+    (med_qa, section side_effects) — never generated advice. Otherwise a plain acknowledgement.
+    """
+    t = load_meds_texts().side_effects
+    yesterday = await checkin_of(ctx, ctx.now.date() - timedelta(days=1))
+    if yesterday is not None and set(codes) & set(yesterday.side_effects or []):
+        return Reply(text=t.repeat, kind="info")
+    rx = await current_prescription(ctx)
+    for code in codes:
+        if code in t.questions and rx is not None:
+            answer = await med_qa(rx.drug, scrub(t.questions[code]), topic="side_effects")
+            if answer is not None:
+                return Reply(text=answer.text, kind="info")
+    return Reply(text=t.ack, kind="info")
+
+
+async def leave_to_idle(ctx: Ctx, event: Event, data: dict[str, Any] | None = None) -> Step:
+    """Abandon the current sub-flow and let the idle state handle the event (menu button, system event...)."""
+    step = Step(next_state="idle", data=data or {})
+    idle = handler_for("idle")
+    return step.then(await idle(ctx, event)) if idle is not None else step
+
+
+async def run_trigger_inline(ctx: Ctx, key: str, event: Event) -> Step | None:
+    """Run an idle trigger from inside another flow (e.g. a reminder button pressed during the check-in)."""
+    handler = get_trigger(key)
+    return await handler(ctx, event) if handler is not None else None
+
+
+# ------------------------------------------------------------------------------------------- reminder
+
+
+def _reminder_buttons(ref: str) -> list[list[Button]]:
+    t = load_meds_texts().reminder
+    return [
+        [Button(label=t.taken, data=f"med:taken:{ref}")],
+        [Button(label=t.skip, data=f"med:skip:{ref}")],
+        [Button(label=t.side_effects, data=f"med:side_effects:{ref}")],
+    ]
+
+
+@trigger("system:med_reminder", replace=True)
+async def _med_reminder(ctx: Ctx, event: Event) -> Step:
+    t = load_meds_texts().reminder
+    rx = await current_prescription(ctx)
+    drug = str(event.params.get("drug") or (rx.drug if rx else ""))
+    dose = str(event.params.get("dose") or (rx.dose if rx else ""))
+    try:
+        dose_id: int | None = int(event.params["dose_id"])  # type: ignore[call-overload]
+    except (KeyError, TypeError, ValueError):
+        dose_id = None
+    if dose_id is None and rx is not None and rx.id is not None:
+        # No scheduler row (demo console without #13): create the planned dose so the answer has a record.
+        row = Dose(token=ctx.token, prescription_id=rx.id, planned_at=ctx.now, status="pending")
+        ctx.session.add(row)
+        await ctx.session.flush()
+        dose_id = row.id
+    text = t.text.format(drug=drug, dose=dose).replace("  ", " ") if drug else t.text_no_drug
+    reply = Reply(text=text, buttons=_reminder_buttons(dose_ref(dose_id, ctx.now.date())), kind="question")
+    return Step(replies=[reply], data={"med_pending": None})  # key of earlier releases: the button carries it now
+
+
+def _stale_reply() -> Step:
+    return Step(replies=[Reply(text=load_meds_texts().stale)])
+
+
+async def _legacy(ctx: Ctx, event: Event) -> Step:
+    """Payload-less medication button of an earlier release: the dose is unknown → treat as stale."""
+    return _stale_reply()
+
+
+def _with_dose(fn: Any) -> Any:
+    """Prefix-trigger handler wrapper: resolve the dose payload, answer stale buttons without recording."""
+
+    async def handler(ctx: Ctx, event: Event) -> Step:
+        ref = trigger_payload(event)
+        fresh, _row = await resolve_ref(ctx, ref)
+        if not fresh:
+            return _stale_reply()
+        return await fn(ctx, event, ref)
+
+    return handler
+
+
+def med_trigger(name: str) -> Any:
+    """Register `button:med:<name>:*` (with the dose payload) plus the legacy payload-less key as stale."""
+
+    def deco(fn: Any) -> Any:
+        trigger(f"button:med:{name}:*")(_with_dose(fn))
+        trigger(f"button:med:{name}")(_legacy)
+        return fn
+
+    return deco
+
+
+@med_trigger("taken")
+async def _taken(ctx: Ctx, event: Event, ref: str) -> Step:
+    await record_dose(ctx, True, ref=ref)
+    return Step(replies=[Reply(text=load_meds_texts().taken_ack)], notify_monitoring=True)
+
+
+@med_trigger("skip")
+async def _skip(ctx: Ctx, event: Event, ref: str) -> Step:
+    await record_dose(ctx, False, status="skipped", ref=ref)
+    t = load_meds_texts().skip
+    buttons = [[Button(label=t.reasons[r], data=f"med:skip_reason:{r}:{ref}")] for r in SKIP_REASONS]
+    return Step(replies=[Reply(text=t.question, buttons=buttons, kind="question")], notify_monitoring=True)
+
+
+def _side_effect_question(ref: str) -> Reply:
+    t = load_meds_texts().side_effects
+    buttons = [[Button(label=t.options[c], data=f"med:se:{c}:{ref}")] for c in SIDE_EFFECT_CODES]
+    return Reply(text=t.question, buttons=buttons, kind="question")
+
+
+def _skip_reason(reason: str) -> Any:
+    async def handler(ctx: Ctx, event: Event, ref: str) -> Step:
+        if reason == "side_effects":
+            return Step(replies=[_side_effect_question(ref)])
+        if reason == "quit":
+            # «Решил бросить» is the same signal as "брошу" in free text: rule `quit_intent` reacts (alert +
+            # delayed-onset explanation through the med_info action).
+            await save_checkin(ctx, quit_intent=True)
+            return Step(replies=[Reply(text=load_meds_texts().skip.ack["quit"])], notify_monitoring=True)
+        return Step(replies=[Reply(text=load_meds_texts().skip.ack[reason])])
+
+    return handler
+
+
+for _reason in SKIP_REASONS:
+    med_trigger(f"skip_reason:{_reason}")(_skip_reason(_reason))
+
+
+@med_trigger("side_effects")
+async def _side_effects(ctx: Ctx, event: Event, ref: str) -> Step:
+    return Step(replies=[_side_effect_question(ref)])
+
+
+async def _after_side_effects(ctx: Ctx, codes: list[str], ref: str | None) -> Step:
+    if ref is None:
+        row = await todays_dose(ctx)
+        ref = dose_ref(row.id if row is not None else None, ctx.now.date())
+    await record_side_effects(ctx, codes, ref=ref)
+    replies = [await side_effect_reply(ctx, codes)]
+    today = await checkin_of(ctx, ctx.now.date())
+    if (today is None or today.med_taken is None) and await current_prescription(ctx) is not None:
+        t = load_meds_texts().side_effects
+        buttons = [
+            [Button(label=t.dose_yes, data=f"med:dose:yes:{ref}")],
+            [Button(label=t.dose_no, data=f"med:dose:no:{ref}")],
+        ]
+        replies.append(Reply(text=t.ask_dose, buttons=buttons, kind="question"))
+    return Step(replies=replies, next_state="idle", data={"med_se_ref": None}, notify_monitoring=True)
+
+
+def _side_effect_code(code: str) -> Any:
+    async def handler(ctx: Ctx, event: Event, ref: str) -> Step:
+        if code == "other":
+            return Step(
+                replies=[Reply(text=load_meds_texts().side_effects.describe)],
+                next_state=SIDE_EFFECT_TEXT_STATE,
+                data={"med_se_ref": ref},
+            )
+        return await _after_side_effects(ctx, [code], ref)
+
+    return handler
+
+
+for _code in SIDE_EFFECT_CODES:
+    med_trigger(f"se:{_code}")(_side_effect_code(_code))
+
+
+@state(SIDE_EFFECT_TEXT_STATE)
+async def _side_effect_text(ctx: Ctx, event: Event) -> Step:
+    if event.type != "text" or ctx.text is None:
+        return await leave_to_idle(ctx, event, {"med_se_ref": None})
+    facts = await ctx.gateway.extract_checkin(ctx.text)
+    # Store only normalised codes, never the description itself (it stays in the scrubbed message log).
+    codes = [str(c) for c in facts.side_effects] if facts is not None and facts.side_effects else ["other"]
+    ref = ctx.data.get("med_se_ref")
+    return await _after_side_effects(ctx, codes, str(ref) if ref else None)
+
+
+def _dose_answer(taken: bool) -> Any:
+    async def handler(ctx: Ctx, event: Event, ref: str) -> Step:
+        await record_dose(ctx, taken, status=None if taken else "missed", ref=ref)
+        return Step(
+            replies=[Reply(text=load_meds_texts().taken_ack if taken else load_meds_texts().skip.ack["other"])],
+            notify_monitoring=True,
+        )
+
+    return handler
+
+
+med_trigger("dose:yes")(_dose_answer(True))
+med_trigger("dose:no")(_dose_answer(False))
+
+# Free-text waiting states of this flow: a long-forgotten question must not swallow tomorrow's message.
+abandonable("meds")
+
+
+# ------------------------------------------------------------------------------------------- question menu
+
+
+@trigger("button:menu:med_question", replace=True)
+async def _menu_question(ctx: Ctx, event: Event) -> Step:
+    t = load_meds_texts().question
+    if await current_prescription(ctx) is None:
+        return Step(replies=[menu_reply(ctx, t.no_prescription)])
+    return Step(replies=[Reply(text=t.prompt)], next_state=QUESTION_STATE)
+
+
+async def answer_question(ctx: Ctx, rx: Rx | None) -> str:
+    """Answer from the instruction excerpt only (SPEC §7 п.5); dose questions and gaps → «уточните у врача»."""
+    if rx is None or ctx.text is None:
+        return load_meds_texts().question.no_prescription
+    answer = await med_qa(rx.drug, ctx.text)
+    return answer.text if answer is not None else fallback_text()
+
+
+@state(QUESTION_STATE)
+async def _question(ctx: Ctx, event: Event) -> Step:
+    if event.type != "text" or ctx.text is None:
+        return await leave_to_idle(ctx, event)
+    text = await answer_question(ctx, await current_prescription(ctx))
+    return Step(replies=[menu_reply(ctx, text)], next_state="idle")
+
+
+# ------------------------------------------------------------------------------------------- rule actions
+
+
+async def med_info_reply(ctx: Ctx, action: Action) -> Reply | None:
+    """`med_info` action of a rule (quit_intent → onset, alcohol_3d → alcohol): instruction excerpt framed by
+    fixed texts. The rule already decided to say it; this only picks the words. None → engine template."""
+    spec = load_meds_texts().med_info.get(str(action.params.get("topic")))
+    if spec is None:
+        return None
+    rx = await current_prescription(ctx)
+    answer = await med_qa(rx.drug, scrub(spec.question), topic=spec.section) if rx is not None else None
+    body = answer.text if answer is not None else spec.fallback
+    return Reply(text="\n\n".join(p for p in (spec.lead, body, spec.notice) if p), kind="info")
+
+
+register_med_info_handler(med_info_reply)

app/dialog/flows/veteran_daily.py

@@ -0,0 +1,465 @@
+"""Daily check-in and scheduled-scale intros (#37, ADR-002, CLINICAL_MODULES §1–2, case «мониторинг после
+обращения»).
+
+Check-in = core (mood 0–10, dose taken) + the module question (veteran: sleep hours + nightmares; alcohol for
+veterans and for anyone with a positive AUDIT-C) + free text:
+
+    system:daily_checkin / menu «Как я сегодня»
+      ──▶ daily.mood        buttons 0–10 (a bare number is accepted too)
+      ──▶ daily.med         only with a prescription and no dose answer today (the reminder may have got it)
+      ──▶ daily.sleep       <4 / 4–6 / 6–8 / >8 h          ┐ module veteran
+      ──▶ daily.nightmares  yes / no                         ┘
+      ──▶ daily.alcohol     none / 1–2 / more  (veteran or AUDIT-C positive)
+      ──▶ daily.text        «Как вы? Пару слов» (skip button) → extract_checkin on the SCRUBBED text
+      ──▶ [daily.alcohol_amount]  alcohol mentioned in the text but not asked by button → amount buttons
+      ──▶ idle              upsert done, `notify_monitoring` → rules (lane B) → dialog actions
+
+Data algorithm (medical data, jury requirement):
+  * Every button answer is written to `daily_checkins` immediately (`repo.upsert_checkin` merges non-None
+    fields), so an interrupted check-in (crisis, the patient walking away) still leaves today's mood on record —
+    the "mood ≤ 3 three days in a row" rule must not miss a day because the free text was never written.
+  * Button answers are the patient's own and always win over the LLM: from the free text the LLM may only add
+    side effects, the quit intent and fields that were not asked by buttons today.
+  * The LLM never decides anything here: alerts and follow-ups (delayed-onset explanation, unscheduled PHQ-9,
+    alcohol interaction, visit offer) come from the rules through `MonitoringService` (ADR-007).
+  * A free-text message classified POSSIBLE by the safety layer is taken over by the engine (C-SSRS). When the
+    screen returns to `daily.text` the message counts as the answer: the check-in is closed, not asked again.
+  * A check-in left unfinished is dropped on the next event of a later day, or after
+    `settings.abandon_after_hours` of silence (engine abandon timeout, #56) — answers given so far stay stored.
+
+Also here: friendly intros for scheduled scales (pcl5, phq9, dar5, auditc), registered with
+`register_scale_intro` so the engine's `scale_due` / rule-triggered sessions show them.
+"""
+
+import re
+from datetime import datetime
+from functools import lru_cache
+from pathlib import Path
+from typing import Any
+
+import yaml
+from pydantic import BaseModel, ConfigDict
+
+from app.config import get_settings
+from app.core.intake_runtime import intake_enabled
+from app.core.rules import load_rules
+from app.db import repo
+from app.dialog.contracts import Button, Event, Reply
+from app.dialog.flows import intake
+from app.dialog.flows.core import menu_reply
+from app.dialog.flows.meds import (
+    answer_question,
+    checkin_of,
+    current_prescription,
+    leave_to_idle,
+    load_meds_texts,
+    record_dose,
+    record_side_effects,
+    run_trigger_inline,
+    save_checkin,
+    side_effect_reply,
+)
+from app.dialog.fsm import Ctx, Step, abandonable, on_enter, state, trigger
+from app.dialog.scale_session import register_scale_intro
+
+PREFIX = "daily"
+STEPS = ("mood", "med", "sleep", "nightmares", "alcohol", "text")
+ALCOHOL_AMOUNT = "alcohol_amount"
+SLEEP_BUCKETS = ("<4", "4-6", "6-8", ">8")
+ALCOHOL_BUCKETS = ("none", "1-2", "more")
+_NUMBER = re.compile(r"^\s*(\d{1,2})\s*$")
+
+
+# ------------------------------------------------------------------------------------------- texts
+
+
+class _Strict(BaseModel):
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+
+class DailySettings(_Strict):
+    sleep_modules: list[str]
+    alcohol_daily_modules: list[str]
+    intro_scales: list[str]
+
+
+class Question(_Strict):
+    question: str
+
+
+class OptionsQuestion(_Strict):
+    question: str
+    options: dict[str, str]
+
+
+class YesNoQuestion(_Strict):
+    question: str
+    yes: str
+    no: str
+
+
+class AlcoholQuestion(_Strict):
+    question: str
+    clarify: str
+    options: dict[str, str]
+
+
+class TextQuestion(_Strict):
+    question: str
+    skip: str
+
+
+class CheckinTexts(_Strict):
+    start: str
+    mood: Question
+    sleep: OptionsQuestion
+    nightmares: YesNoQuestion
+    alcohol: AlcoholQuestion
+    text: TextQuestion
+    buttons_hint: str
+    done: str
+    already_done: str
+
+
+class VisitTexts(_Strict):
+    requested: str
+
+
+class ScaleIntros(_Strict):
+    scheduled: dict[str, str]
+    unscheduled: dict[str, str]
+
+
+class DailyTexts(_Strict):
+    status: str
+    settings: DailySettings
+    checkin: CheckinTexts
+    visit: VisitTexts
+    scale_intros: ScaleIntros
+
+
+def daily_texts_path() -> Path:
+    return get_settings().config_dir / "texts" / "veteran_daily.yaml"
+
+
+@lru_cache
+def load_daily_texts(path: Path | None = None) -> DailyTexts:
+    path = path or daily_texts_path()
+    return DailyTexts.model_validate(yaml.safe_load(path.read_text(encoding="utf-8")))
+
+
+# ------------------------------------------------------------------------------------------- step logic
+
+
+def _today(ctx: Ctx) -> str:
+    return ctx.now.date().isoformat()
+
+
+def _ck(ctx: Ctx) -> dict[str, Any]:
+    return dict(ctx.data.get("checkin") or {})
+
+
+async def _modules(ctx: Ctx) -> set[str]:
+    profile = await ctx.profile()
+    return set(profile.modules or []) if profile is not None else set()
+
+
+async def _auditc_positive(ctx: Ctx) -> bool:
+    """Latest AUDIT-C at or above the sex-specific cut-off of rule `auditc_positive` (one threshold, one place)."""
+    results = await repo.list_scale_results(ctx.session, ctx.token, "auditc")
+    if not results:
+        return False
+    rule = next((r for r in load_rules().rules if r.id == "auditc_positive"), None)
+    if rule is None:
+        return False
+    profile = await ctx.profile()
+    key = "threshold_f" if profile is not None and profile.sex == "f" else "threshold_m"
+    return results[-1].total >= int(rule.params[key])
+
+
+async def _applicable(ctx: Ctx, step: str) -> bool:
+    settings = load_daily_texts().settings
+    if step in ("mood", "text"):
+        return True
+    if step == "med":
+        today = await checkin_of(ctx, ctx.now.date())
+        return (today is None or today.med_taken is None) and await current_prescription(ctx) is not None
+    modules = await _modules(ctx)
+    if step in ("sleep", "nightmares"):
+        return bool(modules & set(settings.sleep_modules))
+    if step == "alcohol":
+        return bool(modules & set(settings.alcohol_daily_modules)) or await _auditc_positive(ctx)
+    return False
+
+
+async def _next_step(ctx: Ctx, after: str | None) -> str:
+    rest = STEPS if after is None else STEPS[STEPS.index(after) + 1 :]
+    for step in rest:
+        if await _applicable(ctx, step):
+            return step
+    return "text"
+
+
+def _buttons(step: str, options: list[tuple[str, str]]) -> list[list[Button]]:
+    return [[Button(label=label, data=f"{PREFIX}:{step}:{value}")] for value, label in options]
+
+
+async def _question(ctx: Ctx, step: str, note: str = "") -> Reply:
+    t = load_daily_texts().checkin
+    buttons: list[list[Button]]
+    if step == "mood":
+        text = t.mood.question
+        # Two rows (0–5, 6–10): 11 buttons in one column are hard to use on a phone.
+        buttons = [
+            [Button(label=str(v), data=f"{PREFIX}:mood:{v}") for v in range(6)],
+            [Button(label=str(v), data=f"{PREFIX}:mood:{v}") for v in range(6, 11)],
+        ]
+    elif step == "med":
+        m = load_meds_texts().checkin
+        rx = await current_prescription(ctx)
+        text = m.question.format(drug=rx.drug) if rx is not None else m.question_no_drug
+        buttons = _buttons(step, [("yes", m.yes), ("no", m.no), ("later", m.later)])
+    elif step == "sleep":
+        text = t.sleep.question
+        buttons = _buttons(step, [(b, t.sleep.options[b]) for b in SLEEP_BUCKETS])
+    elif step == "nightmares":
+        text = t.nightmares.question
+        buttons = _buttons(step, [("yes", t.nightmares.yes), ("no", t.nightmares.no)])
+    elif step in ("alcohol", ALCOHOL_AMOUNT):
+        text = t.alcohol.question if step == "alcohol" else t.alcohol.clarify
+        buttons = _buttons(step, [(b, t.alcohol.options[b]) for b in ALCOHOL_BUCKETS])
+    else:
+        text = t.text.question
+        buttons = _buttons("text", [("skip", t.text.skip)])
+    return Reply(text="\n\n".join(p for p in (note, text) if p), buttons=buttons, kind="question")
+
+
+async def start_checkin(ctx: Ctx, *, from_menu: bool) -> Step:
+    if ctx.data.get("checkin_done") == _today(ctx):
+        # The scheduler may fire twice; the menu button gets a friendly note instead of a second check-in.
+        return Step(replies=[menu_reply(ctx, load_daily_texts().checkin.already_done)]) if from_menu else Step()
+    first = await _next_step(ctx, None)
+    ck = {"day": _today(ctx), "answered": [], "started_at": ctx.now.isoformat()}
+    return Step(
+        replies=[Reply(text=load_daily_texts().checkin.start)],
+        next_state=f"{PREFIX}.{first}",
+        data={"checkin": ck},
+    )
+
+
+def _finish(ctx: Ctx, replies: list[Reply] | None = None) -> Step:
+    done = Reply(text=load_daily_texts().checkin.done)
+    return Step(
+        replies=[done, *(replies or [])],
+        next_state="idle",
+        data={"checkin": None, "checkin_done": _today(ctx)},
+        notify_monitoring=True,
+    )
+
+
+def _consumed_by_safety(ctx: Ctx, ck: dict[str, Any]) -> bool:
+    """The free-text answer was taken over by the safety layer (C-SSRS screen) after the question was asked."""
+    asked = ck.get("text_asked_at")
+    if not asked:
+        return False
+    for event in reversed(ctx.data.get("safety_events") or []):
+        if event.get("state") == f"{PREFIX}.text" and datetime.fromisoformat(event["ts"]) >= datetime.fromisoformat(
+            asked
+        ):
+            return True
+    return False
+
+
+@on_enter(PREFIX)
+async def _enter(ctx: Ctx) -> Step:
+    step = ctx.state.split(".", 1)[1] if "." in ctx.state else ""
+    ck = _ck(ctx)
+    if not ck or ck.get("day") != _today(ctx):
+        return Step(next_state="idle", data={"checkin": None})
+    if step == "text":
+        if _consumed_by_safety(ctx, ck):
+            return _finish(ctx)
+        return Step(
+            replies=[await _question(ctx, step)], data={"checkin": {**ck, "text_asked_at": ctx.now.isoformat()}}
+        )
+    return Step(replies=[await _question(ctx, step)])
+
+
+async def _advance(ctx: Ctx, step: str, ck: dict[str, Any]) -> Step:
+    ck = {**ck, "answered": [*ck.get("answered", []), step]}
+    if step == ALCOHOL_AMOUNT:
+        return _finish(ctx)
+    nxt = await _next_step(ctx, step)
+    return Step(next_state=f"{PREFIX}.{nxt}", data={"checkin": ck})
+
+
+async def _record_button(ctx: Ctx, step: str, value: str) -> bool:
+    """Store one button answer right away. False = value not valid for this step (stale/forged button)."""
+    if step == "mood" and value.isdigit() and 0 <= int(value) <= 10:
+        await save_checkin(ctx, mood=int(value))
+    elif step == "med" and value in ("yes", "no", "later"):
+        if value != "later":  # «Ещё нет» is not a missed dose: nothing is recorded yet
+            await record_dose(ctx, value == "yes", status=None if value == "yes" else "missed")
+    elif step == "sleep" and value in SLEEP_BUCKETS:
+        await save_checkin(ctx, sleep_bucket=value)
+    elif step == "nightmares" and value in ("yes", "no"):
+        await save_checkin(ctx, nightmares=value == "yes")
+    elif step in ("alcohol", ALCOHOL_AMOUNT) and value in ALCOHOL_BUCKETS:
+        await save_checkin(ctx, alcohol_bucket=value)
+    else:
+        return False
+    return True
+
+
+async def _free_text(ctx: Ctx, ck: dict[str, Any]) -> Step:
+    assert ctx.text is not None
+    facts = await ctx.gateway.extract_checkin(ctx.text)
+    answered = set(ck.get("answered", []))
+    fields: dict[str, Any] = {"free_text_scrubbed": ctx.text}
+    replies: list[Reply] = []
+    clarify_alcohol = False
+    if facts is not None:
+        if facts.quit_intent:
+            fields["quit_intent"] = True
+        # Buttons win: the LLM only fills what was not asked today.
+        if "med" not in answered and facts.med_taken is not None:
+            today = await checkin_of(ctx, ctx.now.date())
+            if today is None or today.med_taken is None:
+                fields["med_taken"] = facts.med_taken
+        if "sleep" not in answered and facts.sleep_hours_bucket is not None:
+            fields["sleep_bucket"] = facts.sleep_hours_bucket
+        if "nightmares" not in answered and facts.nightmares is not None:
+            fields["nightmares"] = facts.nightmares
+        if "alcohol" not in answered and facts.alcohol is not None:
+            fields["alcohol_bucket"] = facts.alcohol  # provisional; the amount buttons below refine it
+            clarify_alcohol = facts.alcohol != "none"
+        if facts.side_effects:
+            await record_side_effects(ctx, [str(c) for c in facts.side_effects])
+            # One topic per message: when the text also carries a quit intent or alcohol, the rules answer
+            # those (delayed onset / interaction); the side-effect excerpt would only bury that message.
+            if not facts.quit_intent and facts.alcohol in (None, "none"):
+                replies.append(await side_effect_reply(ctx, [str(c) for c in facts.side_effects]))
+    await save_checkin(ctx, **fields)
+    if clarify_alcohol:
+        return Step(replies=replies, next_state=f"{PREFIX}.{ALCOHOL_AMOUNT}", data={"checkin": ck})
+    return _finish(ctx, replies)
+
+
+abandonable(PREFIX)
+
+
+@state(PREFIX)
+async def _handle(ctx: Ctx, event: Event) -> Step:
+    step = ctx.state.split(".", 1)[1] if "." in ctx.state else ""
+    ck = _ck(ctx)
+    if not ck or ck.get("day") != _today(ctx):
+        # Unfinished check-in from an earlier day: drop it and handle the event as idle would.
+        return await leave_to_idle(ctx, event, {"checkin": None})
+
+    if event.type == "button" and event.button:
+        parts = event.button.split(":")
+        if len(parts) == 3 and parts[0] == PREFIX and parts[1] == step:
+            if step == "text" and parts[2] == "skip":
+                return _finish(ctx)
+            if await _record_button(ctx, step, parts[2]):
+                return await _advance(ctx, step, ck)
+        if parts[0] == "med":
+            # A reminder button pressed mid check-in: answer it, then repeat the current question.
+            inline = await run_trigger_inline(ctx, f"button:{event.button}", event)
+            if inline is not None:
+                return inline if inline.next_state else inline.then(Step(replies=[await _question(ctx, step)]))
+        if parts[0] in (PREFIX, "scale"):
+            return Step(replies=[await _question(ctx, step)])  # stale button of an earlier question
+        return await leave_to_idle(ctx, event, {"checkin": None})
+
+    if event.type == "system" and event.system:
+        if event.system == "daily_checkin":
+            return Step(replies=[await _question(ctx, step)])
+        inline = await run_trigger_inline(ctx, f"system:{event.system}", event)
+        if inline is None:
+            return Step()
+        return inline if inline.next_state else inline.then(Step(replies=[await _question(ctx, step)]))
+
+    if event.type == "text" and ctx.text is not None:
+        if step == "text":
+            return await _free_text(ctx, ck)
+        number = _NUMBER.match(ctx.text)
+        if step == "mood" and number and await _record_button(ctx, "mood", number.group(1)):
+            return await _advance(ctx, step, ck)
+        return Step(replies=[await _question(ctx, step, note=load_daily_texts().checkin.buttons_hint)])
+
+    return Step(replies=[await _question(ctx, step)])
+
+
+# ------------------------------------------------------------------------------------------- idle triggers
+
+
+@trigger("system:daily_checkin", replace=True)
+async def _daily_checkin(ctx: Ctx, event: Event) -> Step:
+    if intake_enabled():  # the intake diary (#53, feature flag) replaces this check-in when released
+        return await intake.checkin(ctx, event)
+    return await start_checkin(ctx, from_menu=False)
+
+
+@trigger("button:menu:checkin", replace=True)
+async def _menu_checkin(ctx: Ctx, event: Event) -> Step:
+    if intake_enabled():
+        return await intake.checkin(ctx, event)
+    return await start_checkin(ctx, from_menu=True)
+
+
+@trigger("button:menu:visit", replace=True)
+async def _visit(ctx: Ctx, event: Event) -> Step:
+    # «Записаться раньше» (rule action suggest_visit): no online booking yet — the request is a fact for lane B
+    # (#38) and the dashboard, like `contact_requests`. No alert is created here (ADR-007).
+    requests = [*(ctx.data.get("visit_requests") or []), {"ts": ctx.now.isoformat()}][-20:]
+    return Step(
+        replies=[Reply(text=load_daily_texts().visit.requested)],
+        data={"visit_requests": requests},
+        notify_monitoring=True,
+    )
+
+
+@trigger("text:idle", replace=True)
+async def _idle_text(ctx: Ctx, event: Event) -> Step:
+    """Free text outside the check-in: drug question → instruction excerpt; side effects / quit intent / alcohol
+    → today's check-in (the rules react); otherwise the neutral support phrase with the menu."""
+    assert ctx.text is not None
+    facts = await ctx.gateway.extract_checkin(ctx.text)
+    if facts is None:
+        return Step(replies=[menu_reply(ctx, await ctx.gateway.phrase("support"))])
+    fields: dict[str, Any] = {}
+    if facts.quit_intent:
+        fields["quit_intent"] = True
+    if facts.alcohol is not None and facts.alcohol != "none":
+        today = await checkin_of(ctx, ctx.now.date())
+        if today is None or today.alcohol_bucket in (None, "none"):
+            fields["alcohol_bucket"] = facts.alcohol
+    text: str | None = None
+    if facts.side_effects:
+        codes = [str(c) for c in facts.side_effects]
+        await record_side_effects(ctx, codes)
+        if not facts.quit_intent:
+            text = (await side_effect_reply(ctx, codes)).text
+    if fields:
+        await save_checkin(ctx, **fields)
+    if text is None and facts.med_question and not facts.quit_intent:
+        text = await answer_question(ctx, await current_prescription(ctx))
+    if text is None:
+        text = await ctx.gateway.phrase("support")
+    return Step(replies=[menu_reply(ctx, text)], notify_monitoring=bool(fields or facts.side_effects))
+
+
+# ------------------------------------------------------------------------------------------- scale intros
+
+
+def _scale_intro(ctx: Ctx, scale_id: str, scheduled: bool, return_state: str) -> str | None:
+    """Friendly intro of a scheduled (or rule-triggered) scale. Onboarding sessions (#36) have their own texts."""
+    if ".onboarding" in return_state:
+        return None
+    intros = load_daily_texts().scale_intros
+    return (intros.scheduled if scheduled else intros.unscheduled).get(scale_id)
+
+
+for _scale_id in load_daily_texts().settings.intro_scales:
+    register_scale_intro(_scale_id, _scale_intro)

app/dialog/flows/veteran_onboarding.py

@@ -0,0 +1,458 @@
+"""Veteran start screening (issue #36, CLINICAL_MODULES §1, ADR-001/002/007/010).
+
+Entered by the core flow after consent + age gate when the profile has module `veteran`:
+
+    veteran.onboarding.start      intro «что нормально после возвращения» + FAQ ──«Начнём»──▶
+    veteran.onboarding.exposure   trauma exposure (yes / no / decline)
+        yes ──▶ scale.pcptsd5 ──▶ ┐
+        no / decline ───────────▶ ┤   (PC-PTSD-5 is not administered; nothing is stored for it)
+    veteran.onboarding.pcl5      ─┘──▶ scale.pcl5
+    veteran.onboarding.phq9        ──▶ scale.phq9   (item 9 buttons only; positive → engine crisis path)
+    veteran.onboarding.auditc      sex from profile, asked if unknown ──▶ scale.auditc
+    veteran.onboarding.weapon      weapon access (4 buttons) → facts; access + risk signal → safe-storage block
+    veteran.onboarding.result      plain-language summary, modules enabled in profile → finish_onboarding()
+
+Clinical decisions encoded here:
+  * PCL-5 is given to every veteran regardless of the exposure answer: the module's start set is PCL-5/PHQ-9/
+    AUDIT-C (CLINICAL_MODULES §1); PC-PTSD-5 is only a short entry screen and, by its own instructions, is asked
+    only after a "yes" to trauma exposure (`config/scales/pcptsd5.yaml: notes`).
+  * Module `depression` is enabled when the PHQ-9 total meets `config/modules.yaml:
+    veteran.enables_depression_if` (≥ 10). Scores and severity labels are never shown to the patient.
+  * Safe-storage block (means restriction, one of the best-evidenced suicide-prevention measures) is shown when
+    the patient reports access to a weapon AND there is a risk signal in this onboarding: a scale flag listed in
+    `settings.weapon_risk_flags` (PHQ-9 item 9 ≥ 1, C-SSRS, AUDIT-C positive), a safety event (POSSIBLE/ACUTE)
+    or a crisis opened since the onboarding began. Deterministic, no LLM (ADR-007).
+
+Facts for lane B (`dialog_state.data["facts"]`, merged with whatever else is there):
+    trauma_exposure        True | False | None (declined)
+    weapon_access          True (self / others_or_safe) | False (none) | None (declined)
+    weapon_access_choice   "none" | "self" | "others_or_safe" | "declined"
+    weapon_access_at       ISO timestamp of the answer
+    weapon_safety_block    True when the safe-storage block was shown
+`PatientSnapshot.weapon_access` should be built from `facts["weapon_access"]`.
+
+Crisis during onboarding: the engine moves to `crisis` and drops the running scale session. Progress is kept in
+`data["veteran_onboarding"]` (`stage`, `since_id`); when the FSM later enters `idle` (doctor acknowledged the
+crisis) the onboarding resumes at the stored stage, and scale stages whose result was already stored since the
+onboarding began are skipped (`since_id` = last scale_results id before the start).
+"""
+
+from __future__ import annotations
+
+from functools import lru_cache
+from pathlib import Path
+from typing import Any
+
+import yaml
+from pydantic import BaseModel, ConfigDict, Field
+
+from app.config import get_settings
+from app.core.modules import load_modules
+from app.db import repo
+from app.db.clinical import ScaleResult
+from app.dialog.contracts import Button, Event, Reply
+from app.dialog.crisis import crisis_open
+from app.dialog.flows.core import finish_onboarding, menu_reply
+from app.dialog.fsm import Ctx, Step, on_enter, state, trigger
+from app.dialog.scale_session import start_scale
+
+MODULE = "veteran"
+PREFIX = "veteran.onboarding"
+START = f"{PREFIX}.start"
+EXPOSURE = f"{PREFIX}.exposure"
+PCL5 = f"{PREFIX}.pcl5"
+PHQ9 = f"{PREFIX}.phq9"
+AUDITC = f"{PREFIX}.auditc"
+WEAPON = f"{PREFIX}.weapon"
+RESULT = f"{PREFIX}.result"
+
+BTN = "vonb"  # button data namespace: "vonb:<question>:<choice>"
+FAQ_KEYS = ("who", "unit", "licences")
+
+# choice code → value of facts["weapon_access"]: storage with others still means the patient owns a weapon.
+WEAPON_ACCESS: dict[str, bool | None] = {"none": False, "self": True, "others_or_safe": True, "declined": None}
+EXPOSURE_VALUE: dict[str, bool | None] = {"yes": True, "no": False, "declined": None}
+SEX_VALUE: dict[str, str | None] = {"m": "m", "f": "f", "declined": None}
+
+
+# ------------------------------------------------------------------------------------------- texts
+
+
+class _Strict(BaseModel):
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+
+class OnbSettings(_Strict):
+    weapon_risk_flags: list[str] = Field(default_factory=list)
+
+
+class IntroTexts(_Strict):
+    text: str
+    go: str
+    repeat: str
+    resume: str
+
+
+class ExposureTexts(_Strict):
+    question: str
+    answer_yes: str
+    answer_no: str
+    decline: str
+    ack_yes: str
+    ack_other: str
+
+
+class SexTexts(_Strict):
+    question: str
+    male: str
+    female: str
+    decline: str
+
+
+class WeaponTexts(_Strict):
+    question: str
+    no_access: str
+    self: str
+    others_or_safe: str
+    decline: str
+    ack: str
+    safety_block_self: str
+    safety_block_stored: str
+    safety_block_unknown: str
+
+
+class ResultTexts(_Strict):
+    intro: str
+    calm: str
+    load: str
+    ptsd: str
+    mood: str
+    alcohol: str
+    next: str
+
+
+class FaqEntry(_Strict):
+    button: str
+    text: str
+
+
+class FaqTexts(_Strict):
+    who: FaqEntry
+    unit: FaqEntry
+    licences: FaqEntry
+
+
+class VeteranOnboardingTexts(_Strict):
+    status: str
+    settings: OnbSettings
+    intro: IntroTexts
+    buttons_hint: str
+    exposure: ExposureTexts
+    sex: SexTexts
+    weapon: WeaponTexts
+    result: ResultTexts
+    faq: FaqTexts
+
+
+def texts_path() -> Path:
+    return get_settings().config_dir / "texts" / "veteran_onboarding.yaml"
+
+
+@lru_cache
+def load_texts(path: Path | None = None) -> VeteranOnboardingTexts:
+    path = path or texts_path()
+    return VeteranOnboardingTexts.model_validate(yaml.safe_load(path.read_text(encoding="utf-8")))
+
+
+# ------------------------------------------------------------------------------------------- helpers
+
+
+def _progress(ctx: Ctx) -> dict[str, Any]:
+    return dict(ctx.data.get("veteran_onboarding") or {})
+
+
+def _at_stage(ctx: Ctx, stage: str) -> dict[str, Any]:
+    """Data update remembering the current stage (resume point after a crisis)."""
+    return {"veteran_onboarding": {**_progress(ctx), "stage": stage}}
+
+
+def _facts(ctx: Ctx, **updates: Any) -> dict[str, Any]:
+    return {"facts": {**(ctx.data.get("facts") or {}), **updates}}
+
+
+def _faq_buttons() -> list[list[Button]]:
+    faq = load_texts().faq
+    return [[Button(label=getattr(faq, key).button, data=f"{BTN}:faq:{key}")] for key in FAQ_KEYS]
+
+
+def _choice(event: Event, question: str, allowed: dict[str, Any]) -> str | None:
+    if event.type != "button" or not event.button:
+        return None
+    parts = event.button.split(":")
+    if len(parts) == 3 and parts[0] == BTN and parts[1] == question and parts[2] in allowed:
+        return parts[2]
+    return None
+
+
+def _faq_key(event: Event) -> str | None:
+    if event.type == "button" and event.button and event.button.startswith(f"{BTN}:faq:"):
+        key = event.button.rsplit(":", 1)[1]
+        return key if key in FAQ_KEYS else None
+    return None
+
+
+async def _results_since(ctx: Ctx) -> list[ScaleResult]:
+    since_id = int(_progress(ctx).get("since_id") or 0)
+    return [r for r in await repo.list_scale_results(ctx.session, ctx.token) if r.id > since_id]
+
+
+async def _latest(ctx: Ctx, scale: str) -> ScaleResult | None:
+    rows = [r for r in await _results_since(ctx) if r.scale == scale]
+    return rows[-1] if rows else None
+
+
+async def _risk_signals(ctx: Ctx) -> list[str]:
+    """Deterministic risk signals collected during this onboarding (see module docstring)."""
+    risk_flags = set(load_texts().settings.weapon_risk_flags)
+    signals = sorted({f for r in await _results_since(ctx) for f in (r.flags or {}) if f in risk_flags})
+    started = str(_progress(ctx).get("started_at") or "")
+    events = [
+        e
+        for e in ctx.data.get("safety_events") or []
+        if str(e.get("ts", "")) >= started and e.get("level") in ("possible", "acute")
+    ]
+    if events:
+        signals.append("safety_event")
+    opened_at = str((ctx.data.get("crisis") or {}).get("opened_at") or "")
+    if crisis_open(ctx.data) or (opened_at and opened_at >= started):
+        signals.append("crisis")
+    return signals
+
+
+async def _scale_stage(ctx: Ctx, stage: str, scale: str, next_stage: str) -> Step:
+    """Start `scale` unless its result is already stored in this onboarding (resume after a crisis)."""
+    if await _latest(ctx, scale) is not None:
+        return Step(next_state=next_stage, data=_at_stage(ctx, next_stage))
+    step = await start_scale(ctx, scale, return_state=next_stage, scheduled=True)
+    step.data.update(_at_stage(ctx, stage))
+    return step
+
+
+# ------------------------------------------------------------------------------------------- start
+
+
+def _intro_reply(text: str) -> Reply:
+    go = Button(label=load_texts().intro.go, data=f"{BTN}:go:yes")
+    return Reply(text=text, buttons=[[go], *_faq_buttons()], kind="question")
+
+
+@on_enter(START)
+async def _start_enter(ctx: Ctx) -> Step:
+    rows = await repo.list_scale_results(ctx.session, ctx.token)
+    progress = {
+        "stage": START,
+        "since_id": max((r.id for r in rows), default=0),
+        "started_at": ctx.now.isoformat(),
+    }
+    return Step(replies=[_intro_reply(load_texts().intro.text)], data={"veteran_onboarding": progress})
+
+
+# ------------------------------------------------------------------------------------------- questions
+
+
+def _exposure_reply(note: str = "") -> Reply:
+    t = load_texts().exposure
+    buttons = [
+        [Button(label=t.answer_yes, data=f"{BTN}:exposure:yes")],
+        [Button(label=t.answer_no, data=f"{BTN}:exposure:no")],
+        [Button(label=t.decline, data=f"{BTN}:exposure:declined")],
+    ]
+    return Reply(text="\n\n".join(p for p in (note, t.question) if p), buttons=buttons, kind="question")
+
+
+def _sex_reply(note: str = "") -> Reply:
+    t = load_texts().sex
+    buttons = [
+        [Button(label=t.male, data=f"{BTN}:sex:m")],
+        [Button(label=t.female, data=f"{BTN}:sex:f")],
+        [Button(label=t.decline, data=f"{BTN}:sex:declined")],
+    ]
+    return Reply(text="\n\n".join(p for p in (note, t.question) if p), buttons=buttons, kind="question")
+
+
+def _weapon_reply(note: str = "") -> Reply:
+    t = load_texts().weapon
+    buttons = [
+        [Button(label=t.no_access, data=f"{BTN}:weapon:none")],
+        [Button(label=t.self, data=f"{BTN}:weapon:self")],
+        [Button(label=t.others_or_safe, data=f"{BTN}:weapon:others_or_safe")],
+        [Button(label=t.decline, data=f"{BTN}:weapon:declined")],
+    ]
+    return Reply(text="\n\n".join(p for p in (note, t.question) if p), buttons=buttons, kind="question")
+
+
+def _current_question(ctx: Ctx, note: str = "") -> Reply:
+    if ctx.state == EXPOSURE:
+        return _exposure_reply(note)
+    if ctx.state == AUDITC:
+        return _sex_reply(note)
+    if ctx.state == WEAPON:
+        return _weapon_reply(note)
+    t = load_texts().intro
+    return _intro_reply("\n\n".join(p for p in (note, t.repeat) if p))
+
+
+@on_enter(EXPOSURE)
+async def _exposure_enter(ctx: Ctx) -> Step:
+    return Step(replies=[_exposure_reply()], data=_at_stage(ctx, EXPOSURE))
+
+
+@on_enter(PCL5)
+async def _pcl5_enter(ctx: Ctx) -> Step:
+    return await _scale_stage(ctx, PCL5, "pcl5", PHQ9)
+
+
+@on_enter(PHQ9)
+async def _phq9_enter(ctx: Ctx) -> Step:
+    # Item 9 is buttons-only and a positive answer opens the crisis path inside the scale session (engine).
+    return await _scale_stage(ctx, PHQ9, "phq9", AUDITC)
+
+
+@on_enter(AUDITC)
+async def _auditc_enter(ctx: Ctx) -> Step:
+    if await _latest(ctx, "auditc") is not None:
+        return Step(next_state=WEAPON, data=_at_stage(ctx, WEAPON))
+    profile = await ctx.profile()
+    known = profile is not None and profile.sex in ("m", "f")
+    if known or _progress(ctx).get("sex_asked"):
+        # Unknown sex after "не хочу отвечать": AUDIT-C uses the more sensitive default threshold.
+        return await _scale_stage(ctx, AUDITC, "auditc", WEAPON)
+    return Step(replies=[_sex_reply()], data=_at_stage(ctx, AUDITC))
+
+
+@on_enter(WEAPON)
+async def _weapon_enter(ctx: Ctx) -> Step:
+    return Step(replies=[_weapon_reply()], data=_at_stage(ctx, WEAPON))
+
+
+@state(PREFIX)
+async def _handle(ctx: Ctx, event: Event) -> Step:
+    """Events in the onboarding's own question states (scales are handled by the scale session)."""
+    t = load_texts()
+    faq = _faq_key(event)
+    if faq is not None:
+        return Step(replies=[Reply(text=getattr(t.faq, faq).text), _current_question(ctx)])
+    if event.type == "system":
+        return Step()  # reminders do not interrupt the screening
+
+    if ctx.state == START:
+        if _choice(event, "go", {"yes": True}):
+            return Step(next_state=EXPOSURE)
+        return Step(replies=[_current_question(ctx)])
+
+    if ctx.state == EXPOSURE and (code := _choice(event, "exposure", EXPOSURE_VALUE)):
+        exposed = EXPOSURE_VALUE[code]
+        step = Step(data=_facts(ctx, trauma_exposure=exposed))
+        if exposed:
+            step.replies.append(Reply(text=t.exposure.ack_yes))
+            nxt = await start_scale(ctx, "pcptsd5", return_state=PCL5, scheduled=True)
+            nxt.data.update(_at_stage(ctx, EXPOSURE))
+            return step.then(nxt)
+        step.replies.append(Reply(text=t.exposure.ack_other))
+        return step.then(Step(next_state=PCL5))
+
+    if ctx.state == AUDITC and (code := _choice(event, "sex", SEX_VALUE)):
+        sex = SEX_VALUE[code]
+        if sex is not None:
+            await repo.upsert_profile(ctx.session, token=ctx.token, sex=sex)
+        step = await _scale_stage(ctx, AUDITC, "auditc", WEAPON)
+        step.data["veteran_onboarding"] = {**step.data["veteran_onboarding"], "sex_asked": True}
+        return step
+
+    if ctx.state == WEAPON and (code := _choice(event, "weapon", WEAPON_ACCESS)):
+        return await _weapon_answer(ctx, code)
+
+    return Step(replies=[_current_question(ctx, t.buttons_hint)])
+
+
+async def _weapon_answer(ctx: Ctx, code: str) -> Step:
+    t = load_texts().weapon
+    access = WEAPON_ACCESS[code]
+    replies = [Reply(text=t.ack)]
+    show_block = access is not False and bool(await _risk_signals(ctx))
+    if show_block:
+        block = {"self": t.safety_block_self, "others_or_safe": t.safety_block_stored}.get(code, t.safety_block_unknown)
+        replies = [Reply(text=block, kind="safety_plan")]
+    data = _facts(
+        ctx,
+        weapon_access=access,
+        weapon_access_choice=code,
+        weapon_access_at=ctx.now.isoformat(),
+        weapon_safety_block=show_block,
+    )
+    # notify_monitoring: weapon access is a monitoring fact (lane B rule "доступ к оружию → блок безопасности").
+    return Step(replies=replies, next_state=RESULT, data=data, notify_monitoring=True)
+
+
+# ------------------------------------------------------------------------------------------- result
+
+
+def _meets(total: int, op: str, value: int) -> bool:
+    return total >= value if op == ">=" else total > value
+
+
+@on_enter(RESULT)
+async def _result_enter(ctx: Ctx) -> Step:
+    t = load_texts().result
+    pcl5 = await _latest(ctx, "pcl5")
+    auditc = await _latest(ctx, "auditc")
+
+    # Depression module: PHQ-9 total vs. the threshold from config/modules.yaml (veteran.enables_depression_if).
+    rule = load_modules().modules[MODULE].enables_depression_if
+    trigger_result = await _latest(ctx, rule.scale) if rule is not None else None
+    depression = bool(rule and trigger_result and _meets(trigger_result.total, rule.op, rule.value))
+    profile = await ctx.profile()
+    modules = list(profile.modules or []) if profile is not None else []
+    for module in (MODULE, "depression") if depression else (MODULE,):
+        if module not in modules:
+            modules.append(module)
+    await repo.upsert_profile(ctx.session, token=ctx.token, modules=modules)
+
+    lines = []
+    if pcl5 is not None and "pcl5_probable_ptsd" in (pcl5.flags or {}):
+        lines.append(t.ptsd)
+    if depression:
+        lines.append(t.mood)
+    if auditc is not None and "auditc_positive" in (auditc.flags or {}):
+        lines.append(t.alcohol)
+    body = [t.intro, t.load, *lines] if lines else [t.intro, t.calm]
+    summary = Reply(text="\n".join([*body, "", t.next]), buttons=_faq_buttons())
+    done = Step(
+        replies=[summary],
+        data={"veteran_onboarding": {**_progress(ctx), "stage": None, "done_at": ctx.now.isoformat()}},
+        notify_monitoring=True,
+    )
+    return done.then(await finish_onboarding(ctx, module=MODULE))
+
+
+# ------------------------------------------------------------------------------------------- resume / FAQ
+
+
+@on_enter("idle")
+async def _resume_after_crisis(ctx: Ctx) -> Step:
+    """Back in idle with an unfinished veteran onboarding (it was interrupted by a crisis) → continue it."""
+    stage = _progress(ctx).get("stage")
+    if not stage or not str(stage).startswith(PREFIX) or crisis_open(ctx.data) or ctx.data.get("consent") is not True:
+        return Step()
+    stage = EXPOSURE if stage == START else str(stage)
+    return Step(replies=[Reply(text=load_texts().intro.resume)], next_state=stage)
+
+
+def _faq_trigger(key: str) -> None:
+    @trigger(f"button:{BTN}:faq:{key}")
+    async def _faq(ctx: Ctx, event: Event) -> Step:
+        return Step(replies=[menu_reply(ctx, getattr(load_texts().faq, key).text)])
+
+
+for _key in FAQ_KEYS:
+    _faq_trigger(_key)

app/dialog/fsm.py

@@ -22,6 +22,14 @@ Name resolution: an exact registration wins, otherwise the longest registered do
 `@trigger("system:daily_checkin")`, `@trigger("text:idle")`) let flows take over what the core `idle` state does
 with menu buttons, system events and free text; the core registers defaults, a flow overrides with
 `replace=True`.
+
+Prefix triggers carry data in the button (#56): `@trigger("button:med:taken:*")` handles every button whose data
+starts with "med:taken:"; the rest of the data ("med:taken:42" → "42") reaches the handler as
+`trigger_payload(event)`. An exact key always wins over a prefix, a longer prefix over a shorter one.
+
+Abandoned steps (#56): a flow marks its question states with `abandonable("daily")`; if the patient stays silent
+in such a state longer than `settings.abandon_after_hours`, the engine returns the FSM to `idle` before
+handling the next event (answers already stored stay stored).
 """
 
 from __future__ import annotations
@@ -103,6 +111,11 @@ EnterHook = Callable[[Ctx], Awaitable[Step]]
 _handlers: dict[str, Handler] = {}
 _enter_hooks: dict[str, EnterHook] = {}
 _triggers: dict[str, Handler] = {}
+_prefix_triggers: dict[str, Handler] = {}  # "button:med:taken:" → handler (registered as "button:med:taken:*")
+_abandonable: dict[str, Callable[[Ctx], bool] | None] = {}
+
+PREFIX_WILDCARD = "*"
+PAYLOAD_PARAM = "trigger_payload"
 
 
 def state(name: str) -> Callable[[Handler], Handler]:
@@ -133,18 +146,45 @@ def trigger(key: str, *, replace: bool = False) -> Callable[[Handler], Handler]:
     """Register a handler for an idle-menu button / system event / idle free text.
 
     Keys: "button:<Button.data>", "system:<SystemEventKind>", "text:idle". The core registers defaults;
-    flows pass `replace=True` to take one over.
+    flows pass `replace=True` to take one over. A key ending in ":*" is a prefix trigger (see module doc).
     """
+    table, name = _triggers, key
+    if key.endswith(":" + PREFIX_WILDCARD):
+        table, name = _prefix_triggers, key[: -len(PREFIX_WILDCARD)]
 
     def deco(fn: Handler) -> Handler:
-        if key in _triggers and not replace:
+        if name in table and not replace:
             raise ValueError(f"trigger '{key}' already registered (use replace=True to override)")
-        _triggers[key] = fn
+        table[name] = fn
         return fn
 
     return deco
 
 
+def trigger_payload(event: Event) -> str | None:
+    """Data after the matched prefix of a prefix trigger ("med:taken:42" → "42"); None for exact triggers."""
+    value = event.params.get(PAYLOAD_PARAM)
+    return str(value) if value is not None else None
+
+
+def abandonable(prefix: str, when: Callable[[Ctx], bool] | None = None) -> None:
+    """Mark states `prefix` / `prefix.*` as question steps the engine may abandon after a long silence.
+
+    `when(ctx)` may veto it for a particular situation (e.g. a scale that is part of an onboarding).
+    """
+    if prefix in _abandonable:
+        raise ValueError(f"abandonable '{prefix}' already registered")
+    _abandonable[prefix] = when
+
+
+def is_abandonable(ctx: Ctx) -> bool:
+    key = _resolve(ctx.state, _abandonable)
+    if key is None:
+        return False
+    when = _abandonable[key]
+    return when is None or when(ctx)
+
+
 def _resolve(name: str, table: dict[str, Any]) -> str | None:
     candidate = name
     while candidate:
@@ -167,7 +207,20 @@ def enter_hook_for(name: str) -> EnterHook | None:
 
 
 def get_trigger(key: str) -> Handler | None:
-    return _triggers.get(key)
+    """Exact trigger, else the longest matching prefix trigger (wrapped so it sees the payload)."""
+    exact = _triggers.get(key)
+    if exact is not None:
+        return exact
+    matches = [p for p in _prefix_triggers if key.startswith(p) and len(key) > len(p)]
+    if not matches:
+        return None
+    prefix = max(matches, key=len)
+    fn, payload = _prefix_triggers[prefix], key[len(prefix) :]
+
+    async def with_payload(ctx: Ctx, event: Event) -> Step:
+        return await fn(ctx, event.model_copy(update={"params": {**event.params, PAYLOAD_PARAM: payload}}))
+
+    return with_payload
 
 
 def is_registered(name: str) -> bool:

app/dialog/scale_session.py

@@ -15,13 +15,22 @@ Algorithm per item:
   4. Every answer is stored immediately (`item_responses`), so a session interrupted by a crisis keeps its data.
   5. After each answer the scale is scored deterministically (`app.core.scales.score`). If a crisis flag from
      `config/texts/dialog.yaml: settings.crisis_flags` fires (PHQ-9 item 9 ≥ 1, C-SSRS high) the session ends at
-     once: partial result saved, fixed crisis script, C-SSRS queued (unless this *is* the C-SSRS). A positive
+     once: partial result saved, fixed crisis script, C-SSRS queued (unless this *is* the C-SSRS, or a C-SSRS was
-     item 9 is acted upon even in a half-finished questionnaire.
+     completed less than `settings.cssrs_cooldown_hours` ago — the patient is not asked the same suicide screen
+     twice in a day; the crisis script and the alert still happen). A positive item 9 is acted upon even in a
+     half-finished questionnaire.
   6. On completion: `scale_results` row (total, severity, item 9, scheduled flag, flags) → the engine commits
      and calls `MonitoringService.after_data_change()` (rules decide alerts — ADR-007) → back to `return_state`.
 The patient never sees a score or a severity label: that would read as a diagnosis.
+
+Intros (#56): `start_scale(intro=None)` looks the intro up in a registry filled by flows
+(`register_scale_intro("phq9", fn_or_text)`), so scales started by the engine (`scale_due`, rule actions) get
+the flow's friendly intro without the flow hooking `scale.<id>` itself.
 """
 
+from collections.abc import Callable
+from dataclasses import replace
+from datetime import timedelta
 from typing import Any
 
 from app.core.privacy import scrub
@@ -29,10 +38,28 @@ from app.core.scales import Item, ScaleDef, ScaleResult, load_scale, score
 from app.db import repo
 from app.dialog.contracts import Button, Event, Reply
 from app.dialog.crisis import crisis_open, open_crisis
-from app.dialog.fsm import Ctx, Step, on_enter, state
+from app.dialog.fsm import RESTING_STATES, Ctx, Step, abandonable, is_abandonable, on_enter, state
 
 PREFIX = "scale"
 
+# (ctx, scale_id, scheduled, return_state) → intro text or None (no intro in this situation).
+IntroFn = Callable[[Ctx, str, bool, str], str | None]
+_intros: dict[str, str | IntroFn] = {}
+
+
+def register_scale_intro(scale_id: str, intro: str | IntroFn, *, replace: bool = False) -> None:
+    """Register the intro shown before `scale_id` when a session is started without an explicit `intro`."""
+    if scale_id in _intros and not replace:
+        raise ValueError(f"intro of scale '{scale_id}' already registered (use replace=True to override)")
+    _intros[scale_id] = intro
+
+
+def scale_intro(ctx: Ctx, scale_id: str, *, scheduled: bool, return_state: str) -> str | None:
+    intro = _intros.get(scale_id)
+    if intro is None or isinstance(intro, str):
+        return intro or None
+    return intro(ctx, scale_id, scheduled, return_state) or None
+
 
 def scale_state(scale_id: str) -> str:
     return f"{PREFIX}.{scale_id}"
@@ -56,9 +83,11 @@ async def start_scale(
     """Begin a session of `scale_id`; returns a Step moving the FSM to `scale.<id>`.
 
     `scheduled=False` marks an unscheduled administration (triggered by a rule or a safety signal).
-    `intro` — optional extra text shown before the scale instruction.
+    `intro` — optional extra text shown before the scale instruction; None → the registered intro, if any.
     """
     scale = load_scale(scale_id)  # fail fast on an unknown id
+    if intro is None:
+        intro = scale_intro(ctx, scale_id, scheduled=scheduled, return_state=return_state)
     sessions = dict(ctx.data.get("scales") or {})
     sessions[scale_id] = {
         "idx": 0,
@@ -98,6 +127,28 @@ def _session(ctx: Ctx) -> tuple[ScaleDef, dict[str, Any]] | None:
     return load_scale(scale_id), session
 
 
+def _abandon_allowed(ctx: Ctx) -> bool:
+    """A silent scale session may be abandoned only if what it returns to may be (idle, a check-in step) —
+    never an onboarding chain, which would otherwise be skipped."""
+    found = _session(ctx)
+    if found is None:
+        return True
+    back = str(found[1].get("return_state") or "idle")
+    if back in RESTING_STATES:
+        return True
+    return back != ctx.state and is_abandonable(replace(ctx, state=back))
+
+
+abandonable(PREFIX, _abandon_allowed)
+
+
+async def recent_risk_screen(ctx: Ctx, screen: str) -> bool:
+    """A complete risk screen (C-SSRS) less than `cssrs_cooldown_hours` ago (scale_results, any trigger)."""
+    since = ctx.now - timedelta(hours=ctx.texts.settings.cssrs_cooldown_hours)
+    results = await repo.list_scale_results(ctx.session, ctx.token, screen)
+    return any(r.complete and r.ts >= since for r in results)
+
+
 @on_enter(PREFIX)
 async def _enter(ctx: Ctx) -> Step:
     found = _session(ctx)
@@ -231,6 +282,7 @@ async def _finish(
     else:
         crisis = open_crisis(ctx, categories=[], source=f"scale:{scale.id}")
     step = done.then(crisis)
-    if scale.id != risk_screen:
+    # Same-day repeat guard: a C-SSRS completed within the cooldown already answered the screening question.
+    if scale.id != risk_screen and not await recent_risk_screen(ctx, risk_screen):
         step.data["pending_scales"] = add_pending(ctx.data, risk_screen, scheduled=False)
     return step

app/dialog/texts.py

@@ -19,6 +19,7 @@ class DialogSettings(_Strict):
     max_safety_events: int = 50
     crisis_flags: list[str] = Field(default_factory=list)
     risk_screen_scale: str = "cssrs_screen"
+    abandon_after_hours: float = 6
 
 
 class Commands(_Strict):

app/scheduler/demo.py

@@ -0,0 +1,257 @@
+"""Demo scheduler on the virtual clock (#13, SPEC §8, ADR-005).
+
+`run_until(now)` replays, in time order, every event that is due by `now` and still fresh
+(`config/scheduler.yaml`):
+
+  * med_reminder  — a `doses` row (pending) per prescription time, then `system:med_reminder {dose_id, drug, dose}`;
+  * missed        — a reminded dose still pending `missed_after_hours` later → status `missed` → monitoring;
+  * daily_checkin — `system:daily_checkin` at the check-in time;
+  * scale_due     — scales due by the module schedules (`config/modules.yaml`) → `system:scale_due`;
+  * evaluation    — once a day `MonitoringService.after_data_change()`: catches silence > N days (rule
+                    `no_contact_3d`) and missed doses even when the patient writes nothing.
+
+The rules decide (ADR-007); the scheduler only turns their dialog actions into system events for the engine
+(`reach_out` → `system:reach_out`, `schedule_scale` → `system:scale_due`). Replies go to `integrations.outbox`.
+
+Idempotent: a delivered event is stored in `scheduler_events` under a unique key and never sent again; a failed
+delivery is not stored and is retried on the next run while it is fresh (at least once). Only patients with
+consent (and adult) are contacted; while a crisis is open no reminders, check-ins or scales are sent.
+Time comes only from the arguments / `app.core.clock` (guardrail #4).
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+from datetime import date, datetime, time, timedelta
+from functools import lru_cache
+from pathlib import Path
+from typing import Any
+
+import yaml
+from pydantic import BaseModel, ConfigDict, Field
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.config import get_settings
+from app.core.modules import due_scales, load_modules
+from app.core.rules import Action
+from app.db import repo
+from app.db.clinical import DialogState, Dose, PatientProfile, Prescription
+from app.db.scheduler import SchedulerEvent
+from app.db.session import get_clinical_session
+from app.dialog.contracts import Event, Reply, get_engine
+from app.integrations.base import integrations
+from app.services.contracts import get_monitoring
+
+log = logging.getLogger(__name__)
+
+# ------------------------------------------------------------------------------------------------ config
+
+
+class _Strict(BaseModel):
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+
+class DosesConfig(_Strict):
+    late_hours: float = Field(gt=0)
+    missed_after_hours: float = Field(gt=0)
+
+
+class Window(_Strict):
+    time: time
+    late_hours: float = Field(gt=0)
+
+
+class SchedulerConfig(_Strict):
+    enabled: bool = True
+    tick_seconds: float = Field(gt=0)
+    doses: DosesConfig
+    daily_checkin: Window
+    scales: Window
+    evaluation: Window
+
+
+@lru_cache
+def load_config(path: Path | None = None) -> SchedulerConfig:
+    path = path or get_settings().config_dir / "scheduler.yaml"
+    return SchedulerConfig.model_validate(yaml.safe_load(path.read_text(encoding="utf-8")))
+
+
+# ------------------------------------------------------------------------------------------------ jobs
+
+# Same due time → this order: the dose before the check-in that asks about it.
+_ORDER = {"evaluation": 0, "missed": 1, "med_reminder": 2, "scale_due": 3, "daily_checkin": 4}
+
+
+@dataclass(order=True)
+class Job:
+    due: datetime
+    rank: int
+    token: str
+    key: str
+    kind: str = field(compare=False)
+    params: dict[str, Any] = field(default_factory=dict, compare=False)
+
+
+def _job(due: datetime, token: str, key: str, kind: str, **params: Any) -> Job:
+    return Job(due=due, rank=_ORDER[kind], token=token, key=key, kind=kind, params=params)
+
+
+def _at(day: date, t: time, now: datetime) -> datetime:
+    return datetime.combine(day, t, tzinfo=now.tzinfo)
+
+
+def _fresh(due: datetime, now: datetime, late_hours: float) -> bool:
+    return due <= now < due + timedelta(hours=late_hours)
+
+
+def _window(day: date, w: Window, now: datetime) -> datetime | None:
+    due = _at(day, w.time, now)
+    return due if _fresh(due, now, w.late_hours) else None
+
+
+async def _dose_jobs(s: AsyncSession, p: PatientProfile, now: datetime, cfg: SchedulerConfig) -> list[Job]:
+    jobs: list[Job] = []
+    today = now.date()
+    for rx in (await s.scalars(select(Prescription).where(Prescription.token == p.token))).all():
+        start = rx.start_date or p.therapy_start
+        for day in (today - timedelta(days=1), today):  # a fresh dose is never older than a day
+            if (start and day < start) or (rx.end_date and day > rx.end_date):
+                continue
+            for raw in (rx.schedule or {}).get("times", []):
+                planned = _at(day, time.fromisoformat(raw), now)
+                if _fresh(planned, now, cfg.doses.late_hours):
+                    key = f"dose:{rx.id}:{planned.isoformat(timespec='minutes')}"
+                    jobs.append(_job(planned, p.token, key, "med_reminder", rx_id=rx.id, planned_at=planned))
+    return jobs
+
+
+async def _missed_jobs(s: AsyncSession, token: str, now: datetime, cfg: SchedulerConfig) -> list[Job]:
+    after = timedelta(hours=cfg.doses.missed_after_hours)
+    stmt = select(Dose).where(Dose.token == token, Dose.status == "pending", Dose.planned_at <= now - after)
+    return [_job(d.planned_at + after, token, f"missed:{d.id}", "missed", dose_id=d.id) for d in await s.scalars(stmt)]
+
+
+async def _scale_jobs(s: AsyncSession, p: PatientProfile, now: datetime, cfg: SchedulerConfig) -> list[Job]:
+    due_at = _window(now.date(), cfg.scales, now)
+    if due_at is None:
+        return []
+    last: dict[str, date] = {}
+    for r in await repo.list_scale_results(s, p.token):
+        if r.complete:
+            last[r.scale] = r.ts.astimezone(now.tzinfo).date()
+    if p.therapy_start:
+        # A scale never taken counts from the visit where therapy started, not as overdue on day 0.
+        for mod in load_modules().modules.values():
+            for entry in mod.schedule:
+                last.setdefault(entry.scale, p.therapy_start)
+    therapy_day = (now.date() - p.therapy_start).days if p.therapy_start else 0
+    scales = due_scales(set(p.modules or []), therapy_day, last, now.date())
+    return [_job(due_at, p.token, f"scale:{sc}:{now.date()}", "scale_due", scale=sc) for sc in scales]
+
+
+async def collect(now: datetime, cfg: SchedulerConfig | None = None) -> list[Job]:
+    """Due and fresh, not yet delivered events of all contactable patients, in time order."""
+    cfg = cfg or load_config()
+    jobs: list[Job] = []
+    async with get_clinical_session() as s:
+        for p in (await s.scalars(select(PatientProfile).order_by(PatientProfile.token))).all():
+            state = await s.get(DialogState, p.token)
+            data = state.data if state is not None else {}
+            if data.get("consent") is not True or data.get("adult") is not True:
+                continue
+            mine = await _missed_jobs(s, p.token, now, cfg)
+            if (due := _window(now.date(), cfg.evaluation, now)) is not None:
+                mine.append(_job(due, p.token, f"eval:{now.date()}", "evaluation"))
+            if not (data.get("crisis") or {}).get("open"):
+                mine += await _dose_jobs(s, p, now, cfg)
+                mine += await _scale_jobs(s, p, now, cfg)
+                if (due := _window(now.date(), cfg.daily_checkin, now)) is not None:
+                    mine.append(_job(due, p.token, f"checkin:{now.date()}", "daily_checkin"))
+            done = set((await s.scalars(select(SchedulerEvent.key).where(SchedulerEvent.token == p.token))).all())
+            jobs += [j for j in mine if j.key not in done]
+    return sorted(jobs)
+
+
+# ------------------------------------------------------------------------------------------------ delivery
+
+
+async def _system(token: str, kind: str, **params: Any) -> None:
+    replies: list[Reply] = await get_engine().handle(
+        token,
+        Event(type="system", system=kind, params=params),  # type: ignore[arg-type]
+    )
+    if replies:
+        await integrations.outbox.push(token, replies)
+
+
+async def _after_rules(token: str) -> None:
+    """Monitoring pass outside a dialog: hand the rules' dialog actions to the engine as system events."""
+    actions: list[Action] = await get_monitoring().after_data_change(token)
+    for a in actions:
+        if a.type == "reach_out":
+            await _system(token, "reach_out", **a.params)
+        elif a.type == "schedule_scale" and a.params.get("scale"):
+            await _system(token, "scale_due", scale=a.params["scale"], scheduled=bool(a.params.get("scheduled")))
+        elif a.type == "crisis":
+            log.warning("scheduler crisis action token=%s code=%s", token, a.code)  # alert already persisted
+
+
+async def _ensure_dose(token: str, rx_id: int, planned_at: datetime) -> tuple[int, Prescription]:
+    async with get_clinical_session() as s:
+        rx = await s.get(Prescription, rx_id)
+        assert rx is not None
+        dose = await s.scalar(
+            select(Dose).where(Dose.token == token, Dose.prescription_id == rx_id, Dose.planned_at == planned_at)
+        )
+        if dose is None:
+            dose = Dose(token=token, prescription_id=rx_id, planned_at=planned_at, status="pending")
+            s.add(dose)
+            await s.flush()
+        return dose.id, rx
+
+
+async def _mark_missed(dose_id: int) -> bool:
+    async with get_clinical_session() as s:
+        dose = await s.get(Dose, dose_id)
+        if dose is None or dose.status != "pending":
+            return False  # answered meanwhile
+        dose.status = "missed"
+        return True
+
+
+async def _run(job: Job, now: datetime) -> None:
+    if job.kind == "med_reminder":
+        dose_id, rx = await _ensure_dose(job.token, job.params["rx_id"], job.params["planned_at"])
+        await _system(job.token, "med_reminder", dose_id=dose_id, drug=rx.drug, dose=rx.dose)
+    elif job.kind == "missed":
+        if await _mark_missed(job.params["dose_id"]):
+            await _after_rules(job.token)
+    elif job.kind == "daily_checkin":
+        await _system(job.token, "daily_checkin")
+    elif job.kind == "scale_due":
+        await _system(job.token, "scale_due", scale=job.params["scale"], scheduled=True)
+    elif job.kind == "evaluation":
+        await _after_rules(job.token)
+
+
+async def _done(job: Job, now: datetime) -> None:
+    async with get_clinical_session() as s:
+        s.add(SchedulerEvent(token=job.token, key=job.key, kind=job.kind, due_at=job.due, done_at=now))
+
+
+async def run_until(now: datetime, cfg: SchedulerConfig | None = None) -> list[Job]:
+    """Deliver every due event up to `now` in time order; returns the delivered jobs."""
+    delivered: list[Job] = []
+    for job in await collect(now, cfg):
+        try:
+            await _run(job, now)
+        except Exception as exc:  # noqa: BLE001 — one patient's failure must not stop the others
+            log.error("scheduler job failed token=%s kind=%s error=%s", job.token, job.kind, type(exc).__name__)
+            continue
+        await _done(job, now)
+        delivered.append(job)
+    if delivered:
+        log.info("scheduler delivered=%s", [(j.token, j.kind) for j in delivered])
+    return delivered

app/scheduler/jobs.py

@@ -1,4 +1,4 @@
-"""Questionnaire jobs on the demo clock; clinical release disabled by default.
+"""Scheduler on the demo clock: questionnaire jobs when that release is on, else the demo scheduler (#13).
 
 One scheduler per database. PostgreSQL's advisory transaction lock serializes
 workers; an asyncio lock does the same within a SQLite demonstration process.
@@ -12,6 +12,7 @@ from datetime import datetime
 from zoneinfo import ZoneInfo
 
 from sqlalchemy import select, text
+from sqlalchemy.exc import SQLAlchemyError
 
 from app.core import clock
 from app.core.intake_runtime import due_scales, intake_enabled, load_runtime
@@ -21,6 +22,8 @@ from app.db.intake import QuestionnaireTask
 from app.db.session import get_clinical_session
 from app.dialog.contracts import Event, Reply, get_engine
 from app.integrations.base import integrations
+from app.scheduler import demo
+from app.services import monitoring
 from app.services import questionnaire as service
 
 log = logging.getLogger(__name__)
@@ -146,6 +149,9 @@ async def run_until_now():
             return
         await service.dispatch_notices()
         if not intake_enabled():
+            # Questionnaire release off → the demo scheduler (doses, check-ins, scales, silence) runs instead.
+            if demo.load_config().enabled:
+                await demo.run_until(clock.now())
             return
         await plan()
         await deliver()
@@ -158,12 +164,16 @@ async def _run():
             await run_until_now()
         except Exception as exc:
             log.error("questionnaire scheduler error=%s", type(exc).__name__)
-        await asyncio.sleep(30)
+        await asyncio.sleep(demo.load_config().tick_seconds)
 
 
 async def start():
     global _worker
-    if _worker is None and intake_enabled():
+    try:
+        await monitoring.init_clock_persistence()
+    except (OSError, SQLAlchemyError) as exc:
+        log.error("clock persistence unavailable error=%s", type(exc).__name__)
+    if _worker is None and (intake_enabled() or demo.load_config().enabled):
         _worker = asyncio.create_task(_run())
 
 

app/services/monitoring.py

@@ -0,0 +1,157 @@
+"""Monitoring service (#38): patient snapshot → deterministic rules → persisted alerts (ADR-007).
+
+`after_data_change(token)` is called by the dialog engine after it commits clinical data (and by the scheduler
+on silence). It builds a `PatientSnapshot`, runs `app.core.rules.evaluate`, then in one transaction:
+  * writes every `alert`/`crisis` action to `alerts` (a crisis is an `acute` alert) and, for a crisis,
+    an audit record `crisis_acute` (guardrail #6);
+  * applies `enhanced_monitoring` to the profile;
+  * refreshes the cached `status_color` of the profile (ADR-006; the source of truth stays the open alerts).
+After the commit the doctor is notified through `integrations.notifier` (a crisis also goes to the doctor on
+duty), and only the *dialog* actions are returned to the engine — it never creates alerts itself.
+
+A crisis rule whose acute alert is still open stays silent: the crisis script is already running and the
+doctor already knows, so it is neither restarted nor re-alerted (`rules.evaluate` never suppresses crises).
+
+The virtual clock offset is persisted in `clock_state` (ADR-005) by `init_clock_persistence()`.
+"""
+
+import asyncio
+import logging
+from datetime import datetime, timedelta
+
+from sqlalchemy.exc import SQLAlchemyError
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core import clock
+from app.core.rules import Action, evaluate, patient_status
+from app.core.types import AlertLevel
+from app.db import repo
+from app.db.audit import audit
+from app.db.session import get_clinical_session
+from app.integrations.base import integrations
+from app.services.contracts import register_monitoring
+from app.services.snapshot import build_snapshot
+
+log = logging.getLogger(__name__)
+
+SYSTEM_ACTOR = "system"
+# Notification address of the doctor on duty: acute alerts must reach someone even when the attending doctor
+# is off (SPEC: crisis → immediate escalation). The notifier maps it to the duty channel.
+ON_DUTY_DOCTOR = "on-duty"
+# Persisted by the service itself, never passed to the dialog engine.
+_SERVICE_ACTIONS = frozenset({"alert", "enhanced_monitoring"})
+
+
+async def refresh_status(session: AsyncSession, token: str) -> str:
+    """Recompute and cache the dashboard colour from open alerts (also call it after an ack)."""
+    alerts = await repo.list_alerts(session, token=token, open_only=True)
+    status = patient_status(AlertLevel[a.level.upper()] for a in alerts)
+    await repo.upsert_profile(session, token=token, status_color=status.color)
+    return status.color
+
+
+def _drop_running_crises(actions: list[Action], open_acute: set[str]) -> list[Action]:
+    """Silence every action of a crisis rule whose acute alert is still open (see module docstring)."""
+    running = {a.code for a in actions if a.type == "crisis" and a.code in open_acute}
+    return [a for a in actions if a.code not in running]
+
+
+class RuleMonitoring:
+    async def after_data_change(self, token: str) -> list[Action]:
+        now = clock.now()
+        notify: list[tuple[str, str, str]] = []  # (doctor_id, level, text) — sent after the commit
+        async with get_clinical_session() as s:
+            profile = await repo.get_profile(s, token)
+            snapshot = await build_snapshot(s, token, now, profile=profile)
+            if profile is None or snapshot is None:
+                return []
+            open_acute = {a.code for a in await repo.list_alerts(s, token=token, open_only=True) if a.level == "acute"}
+            actions = _drop_running_crises(evaluate(snapshot), open_acute)
+            for action in actions:
+                if action.type == "enhanced_monitoring":
+                    profile.enhanced_monitoring = True
+                if not action.creates_alert or action.level is None:
+                    continue
+                alert = await repo.add_alert(
+                    s,
+                    token=token,
+                    level=action.level,
+                    code=action.code,
+                    reason=action.reason_ru,
+                    ts=now,
+                    params=action.params,
+                )
+                level = action.level.name.lower()
+                notify.append((profile.doctor_id, level, action.reason_ru))
+                if action.type == "crisis":
+                    await audit(
+                        s,
+                        actor=SYSTEM_ACTOR,
+                        action="crisis_acute",
+                        token=token,
+                        details={"alert_id": alert.id, "code": action.code},
+                        ts=now,
+                    )
+                    notify.append((ON_DUTY_DOCTOR, level, action.reason_ru))
+            await refresh_status(s, token)
+        log.info("monitoring token=%s actions=%s", token, [a.code for a in actions])
+        for doctor_id, level, text in notify:
+            await _notify(doctor_id, level, text, token)
+        return [a for a in actions if a.type not in _SERVICE_ACTIONS]
+
+    async def is_in_crisis(self, token: str) -> bool:
+        async with get_clinical_session() as s:
+            alerts = await repo.list_alerts(s, token=token, open_only=True)
+        return any(a.level == "acute" for a in alerts)
+
+
+async def _notify(doctor_id: str, level: str, text: str, token: str) -> None:
+    # The alert is already committed and visible on the dashboard; a failing channel must not undo it.
+    try:
+        await integrations.notifier.send(doctor_id, level, text, token)  # type: ignore[arg-type]
+    except Exception as exc:  # noqa: BLE001 — any channel error
+        log.warning("notify failed token=%s level=%s error=%s", token, level, type(exc).__name__)
+
+
+register_monitoring(RuleMonitoring)
+
+
+# ---------------------------------------------------------------------------------- demo clock persistence
+
+_writes: set[asyncio.Task[None]] = set()
+
+
+async def _write_offset() -> None:
+    # Writes the offset current at write time, so out-of-order tasks still leave the latest value stored.
+    try:
+        async with get_clinical_session() as s:
+            await repo.save_clock_offset(s, clock.offset().total_seconds())
+    except (OSError, SQLAlchemyError) as exc:
+        log.warning("clock offset not saved error=%s", type(exc).__name__)
+
+
+def _save_offset(_: timedelta) -> None:
+    """Sync hook for `clock.register_persistence`: schedules the DB write on the running loop."""
+    try:
+        loop = asyncio.get_running_loop()
+    except RuntimeError:
+        log.warning("clock offset not saved: no running event loop")
+        return
+    task = loop.create_task(_write_offset())
+    _writes.add(task)
+    task.add_done_callback(_writes.discard)
+
+
+async def flush_clock_writes() -> None:
+    """Wait for pending offset writes (tests, shutdown)."""
+    if _writes:
+        await asyncio.gather(*list(_writes))
+
+
+async def init_clock_persistence() -> datetime:
+    """Restore the virtual clock offset from `clock_state` and save every later change there."""
+    async with get_clinical_session() as s:
+        stored = timedelta(seconds=await repo.load_clock_offset(s))
+    clock.register_persistence(lambda: stored, _save_offset)
+    log.info("clock offset restored seconds=%s", int(stored.total_seconds()))
+    return clock.now()

app/services/snapshot.py

@@ -0,0 +1,93 @@
+"""Patient snapshot for the rules engine, assembled from the clinical DB (ADR-007).
+
+Maps storage names to `app.core.rules` names (`sleep_bucket` → `sleep_hours_bucket`, `alcohol_bucket` →
+`alcohol`), derives age from `birth_year` and the therapy day from `therapy_start`. Time comes in as `as_of`
+(from `app.core.clock`), never from the wall clock. Must not import `app.db.identity` (guardrail #5).
+"""
+
+from datetime import datetime, timedelta
+from typing import Any, get_args
+
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.rules import (
+    CssrsRisk,
+    DailyCheckin,
+    DoseRecord,
+    Module,
+    PatientSnapshot,
+    ScalePoint,
+)
+from app.db import repo
+from app.db.clinical import Dose, Message, PatientProfile
+
+# Risk screens are stored as scale results but reach the rules as `cssrs_risk`, not as scale points.
+CSSRS_SCALES = frozenset({"cssrs_screen", "cssrs"})
+# A C-SSRS result counts as the current risk for this long (same window as PHQ-9 item 9 and the dialog's
+# C-SSRS cooldown): one old "high" must not restart the crisis protocol after the doctor has acted on it.
+CSSRS_RECENT = timedelta(hours=24)
+# Unknown sex/age fall back to the more cautious option: lower AUDIT-C cut-off, young-adult monitoring.
+_UNKNOWN_SEX = "f"
+_UNKNOWN_AGE = 0
+
+_MODULES = frozenset(get_args(Module))
+_CSSRS_RISKS = frozenset(get_args(CssrsRisk))
+
+
+def _cssrs_risk(results: list[Any], as_of: datetime) -> CssrsRisk | None:
+    recent = [r for r in results if r.scale in CSSRS_SCALES and as_of - CSSRS_RECENT < r.ts <= as_of]
+    if not recent:
+        return None
+    severity = recent[-1].severity
+    return severity if severity in _CSSRS_RISKS else None
+
+
+async def build_snapshot(
+    session: AsyncSession, token: str, as_of: datetime, *, profile: PatientProfile | None = None
+) -> PatientSnapshot | None:
+    """Snapshot of everything the rules may look at; None if the patient has no clinical profile."""
+    profile = profile or await repo.get_profile(session, token)
+    if profile is None:
+        return None
+
+    checkins = [
+        DailyCheckin(
+            day=c.day,
+            mood=c.mood,
+            med_taken=c.med_taken,
+            sleep_hours_bucket=c.sleep_bucket,  # type: ignore[arg-type]
+            alcohol=c.alcohol_bucket,  # type: ignore[arg-type]
+            side_effects=list(c.side_effects or []),
+            quit_intent=bool(c.quit_intent),
+            nightmares=c.nightmares,
+        )
+        for c in await repo.list_checkins(session, token)
+    ]
+    results = [r for r in await repo.list_scale_results(session, token) if r.complete]
+    doses = (await session.scalars(select(Dose).where(Dose.token == token).order_by(Dose.planned_at))).all()
+    last_in = await session.scalar(
+        select(func.max(Message.ts)).where(Message.token == token, Message.direction == "in")
+    )
+    state = await repo.get_dialog_state(session, token)
+    facts = (state.data.get("facts") or {}) if state is not None else {}
+
+    return PatientSnapshot(
+        token=token,
+        as_of=as_of,
+        age=as_of.year - profile.birth_year if profile.birth_year else _UNKNOWN_AGE,
+        sex=profile.sex if profile.sex in ("m", "f") else _UNKNOWN_SEX,  # type: ignore[arg-type]
+        modules={m for m in profile.modules or [] if m in _MODULES},  # type: ignore[misc]
+        therapy_day=(as_of.date() - profile.therapy_start).days if profile.therapy_start else 0,
+        checkins=checkins,
+        scale_results=[
+            ScalePoint(scale=r.scale, ts=r.ts, total=r.total, item9=r.item9, scheduled=r.scheduled)
+            for r in results
+            if r.scale not in CSSRS_SCALES
+        ],
+        doses=[DoseRecord(planned_at=d.planned_at, status=d.status) for d in doses],  # type: ignore[arg-type]
+        cssrs_risk=_cssrs_risk(results, as_of),
+        weapon_access=facts.get("weapon_access"),
+        last_patient_message_at=last_in,
+        open_alert_codes=await repo.open_alert_codes(session, token),
+    )

config/scheduler.yaml

@@ -0,0 +1,25 @@
+# Demo scheduler on the virtual clock (#13, SPEC §8, ADR-005). Local times are in settings.tz.
+# Read by app/scheduler/demo.py; used only while the questionnaire release (intake_runtime.yaml) is off.
+#
+# Every event is delivered once (idempotency key in `scheduler_events`) and only while it is fresh: an event
+# older than its late window is not sent at all — after a long jump of the demo clock the patient does not
+# receive a pile of yesterday's reminders, and an undelivered reminder never turns into a "missed" dose.
+
+enabled: true
+tick_seconds: 30
+
+doses:
+  late_hours: 2            # reminder not delivered if the dose time is older than this
+  missed_after_hours: 2    # reminded dose without an answer → status "missed" (SPEC §8)
+
+daily_checkin:
+  time: "20:00"
+  late_hours: 4
+
+scales:
+  time: "10:00"            # scales due today (config/modules.yaml schedules) are offered from this hour
+  late_hours: 10
+
+evaluation:
+  time: "09:00"            # daily monitoring pass: silence > N days, missed doses → caring contact
+  late_hours: 15

config/texts/dialog.yaml

@@ -16,6 +16,9 @@ settings:
   crisis_flags: [phq9_item9_positive, cssrs_high]
   # Шкала-скрининг суицидального риска (POSSIBLE и положительный п.9 PHQ-9 → эта шкала).
   risk_screen_scale: cssrs_screen
+  # Брошенный шаг (чек-ин, шкала): если пациент молчит дольше N часов, следующее событие сначала возвращает
+  # диалог в idle (уже данные ответы сохранены), затем обрабатывается. 0 — выключено. На кризис не действует.
+  abandon_after_hours: 6
 
 commands:
   # Текстовые команды отзыва согласия (сравниваются после приведения к нижнему регистру).

config/texts/meds.yaml

@@ -0,0 +1,98 @@
+# Тексты приёма препарата (issue #37): напоминание, пропуск, побочные эффекты, вопрос о лекарстве, ответы
+# на действия правил med_info (отсроченный эффект, алкоголь, побочные эффекты).
+# СТАТУС: черновик, утверждает психиатр (#8).
+#
+# Правила текстов: без диагнозов и названий расстройств; без советов по дозам и схеме приёма (решает врач);
+# не использовать слова, которые звучат как постановка на учёт или слежка (проверяется тестом). Ответы о
+# препарате берутся только из выдержки инструкции ГРЛС (app/llm/med_qa.py); тексты ниже — обрамление и
+# запасные варианты, если выдержки нет.
+
+status: "черновик, утверждает психиатр (#8)"
+
+reminder:
+  # {drug}, {dose} — из назначения врача (МНН и дозировка как в назначении, без изменений).
+  text: "Время принять {drug} {dose}. Отметьте, пожалуйста, как получилось."
+  text_no_drug: "Время принять препарат, назначенный врачом. Отметьте, пожалуйста, как получилось."
+  taken: "Принял(а)"
+  skip: "Пропущу"
+  side_effects: "Есть побочные эффекты"
+
+taken_ack: "Отлично, отметил приём. Спасибо!"
+# Кнопка от напоминания, которое уже неактуально (например, вчерашнего).
+stale: "Это напоминание уже неактуально. Если хотите отметить самочувствие — нажмите «Как я сегодня»."
+
+skip:
+  question: "Понял, отметил. Подскажите, что помешало? Это поможет вашему врачу."
+  reasons:
+    forgot: "Забыл(а)"
+    side_effects: "Из-за побочных эффектов"
+    quit: "Решил(а) бросить"
+    other: "Другое"
+  ack:
+    forgot: "Понял, бывает. Завтра напомню в обычное время."
+    quit: "Спасибо, что честно сказали — это важно, и решать это лучше вместе с врачом."
+    other: "Понял, отметил. Если захотите рассказать подробнее — просто напишите."
+
+side_effects:
+  question: "Что беспокоит? Выберите вариант или «Другое», чтобы описать своими словами."
+  options:
+    nausea: "Тошнота"
+    insomnia: "Бессонница"
+    headache: "Головная боль"
+    dizziness: "Головокружение"
+    agitation: "Беспокойство"
+    other: "Другое"
+  # Короткий вопрос к инструкции препарата (med_qa, раздел «побочные эффекты») для выбранного эффекта.
+  questions:
+    nausea: "тошнота после приёма, когда пройдёт"
+    insomnia: "бессонница, нарушения сна"
+    headache: "головная боль"
+    dizziness: "головокружение"
+    agitation: "беспокойство, тревога"
+  describe: "Опишите, пожалуйста, коротко, что вы заметили."
+  ack: "Спасибо, отметил. Ваш врач увидит это в вашей карточке."
+  # Тот же эффект был и вчера: инструкцию второй раз не повторяем.
+  repeat: "Понимаю, это неприятно. Я отмечаю это каждый день, и ваш врач увидит, если не проходит."
+  ask_dose: "А препарат сегодня приняли?"
+  dose_yes: "Да, принял(а)"
+  dose_no: "Нет"
+
+# Ежедневный чек-ин: если сегодня ещё нет отметки о приёме.
+checkin:
+  question: "Сегодня приняли {drug}?"
+  question_no_drug: "Сегодня приняли назначенный препарат?"
+  "yes": "Да"
+  "no": "Нет"
+  later: "Ещё нет"
+
+question:
+  prompt: "Напишите ваш вопрос о препарате. Я отвечаю только по официальной инструкции, а если ответа в ней нет — подскажу обратиться к врачу."
+  no_prescription: "Я не нашёл в карточке назначенного препарата. Пожалуйста, уточните вопрос у вашего врача."
+
+# Ответы на действие правила med_info. Ответ = lead + выдержка из инструкции (med_qa по разделу section,
+# вопрос question) + notice. Если выдержки нет — lead + fallback + notice.
+med_info:
+  delayed_effect:
+    section: onset
+    question: "когда начнёт действовать препарат, нет эффекта, только хуже"
+    lead: "Понимаю, что сейчас тяжело."
+    fallback: "Эффект таких препаратов обычно наступает не сразу, а через 2–4 недели регулярного приёма. Пожалуйста, не прекращайте приём самостоятельно."
+    notice: "Я сообщил вашему врачу — он свяжется с вами, чтобы обсудить, как вы себя чувствуете."
+  alcohol_interaction:
+    section: alcohol
+    question: "можно ли алкоголь вместе с препаратом, сон"
+    lead: "Спасибо, что честно рассказали."
+    fallback: "В инструкции к препарату есть предупреждение о совместном приёме с алкоголем, а алкоголь сам по себе ухудшает сон."
+    notice: "Я передал это вашему врачу."
+  nausea:
+    section: side_effects
+    question: "тошнота после приёма, когда пройдёт"
+    lead: ""
+    fallback: "Тошнота часто встречается в первые дни приёма и обычно проходит сама."
+    notice: "Если станет хуже — напишите мне, я передам врачу."
+  side_effects:
+    section: side_effects
+    question: "побочные эффекты в начале приёма"
+    lead: ""
+    fallback: "Спасибо, что сообщили о самочувствии. Я отмечу это для вашего врача."
+    notice: "Если станет хуже — напишите мне, я передам врачу."

config/texts/veteran_daily.yaml

@@ -0,0 +1,62 @@
+# Тексты ежедневного чек-ина и плановых шкал (issue #37, ADR-002, CLINICAL_MODULES §1–2).
+# Ядро (настроение + приём) задаётся всем; вопрос про сон и кошмары — модулю veteran; алкоголь — см. settings.
+# СТАТУС: черновик, утверждает психиатр (#8).
+#
+# Правила текстов: без диагнозов и названий расстройств; без советов по дозам; пациент не видит баллов шкал;
+# не использовать слова, которые звучат как постановка на учёт или слежка (проверяется тестом).
+
+status: "черновик, утверждает психиатр (#8)"
+
+settings:
+  # Модули, которым задаётся вопрос про сон и кошмары (modules.yaml: daily_question = sleep_nightmares).
+  sleep_modules: [veteran]
+  # Ежедневный вопрос об алкоголе: модули, где он задаётся всегда (у ветеранов алкоголь — частый способ
+  # «заглушить» бессонницу и фактор риска, CLINICAL_MODULES §1); остальным — при положительном AUDIT-C
+  # (порог берётся из правила auditc_positive в config/rules.yaml, ADR-002).
+  alcohol_daily_modules: [veteran]
+  # Плановые шкалы, перед которыми показывается дружелюбное вступление (см. scale_intros).
+  intro_scales: [pcl5, phq9, dar5, auditc]
+
+checkin:
+  start: "Добрый день! Короткая отметка о самочувствии — это займёт меньше минуты."
+  mood:
+    question: "Как ваше настроение сегодня по шкале от 0 до 10? 0 — очень плохо, 10 — отлично."
+  sleep:
+    question: "Сколько часов удалось поспать прошлой ночью?"
+    options:
+      "<4": "Меньше 4 ч"
+      "4-6": "4–6 ч"
+      "6-8": "6–8 ч"
+      ">8": "Больше 8 ч"
+  nightmares:
+    question: "Снились ли кошмары или тяжёлые сны?"
+    "yes": "Да"
+    "no": "Нет"
+  alcohol:
+    question: "Выпивали ли вы алкоголь за последние сутки?"
+    clarify: "Вы упомянули алкоголь. Сколько примерно выпили за последние сутки?"
+    options:
+      none: "Не пил(а)"
+      "1-2": "1–2 порции"
+      more: "Больше"
+  text:
+    question: "Как вы? Можно пару слов — что было за день, что беспокоит."
+    skip: "Пропустить"
+  buttons_hint: "Выберите, пожалуйста, вариант кнопкой."
+  done: "Спасибо, записал. Если что-то изменится — пишите в любое время."
+  already_done: "Сегодня вы уже отмечались, спасибо! Если хотите что-то добавить — просто напишите."
+
+visit:
+  # Нажата кнопка «Записаться к врачу раньше» (действие правила suggest_visit). Онлайн-записи пока нет —
+  # просьба передаётся врачу как факт в карточке.
+  requested: "Передал врачу просьбу о более раннем визите. С вами свяжутся, чтобы договориться о времени. Если станет хуже до визита — пишите мне или звоните 112."
+
+# Дружелюбное вступление перед шкалой (параметр intro у start_scale). Пациент не видит баллов.
+scale_intros:
+  scheduled:
+    pcl5: "Раз в две недели я прошу ответить на вопросы о том, как беспокоят тяжёлые воспоминания и их последствия. Это займёт 3–4 минуты. Так ваш врач увидит, что меняется со временем."
+    phq9: "Пора для короткого опроса о настроении и силах за последние две недели — 9 вопросов. Отвечайте так, как есть, правильных ответов нет."
+    dar5: "Несколько вопросов о раздражительности и вспышках гнева. После службы это бывает у многих, и врачу важно видеть, как это меняется."
+    auditc: "Три коротких вопроса об алкоголе. Отвечайте как есть: это нужно врачу, чтобы лечение было безопасным."
+  unscheduled:
+    phq9: "Последние дни настроение ниже обычного. Давайте я задам несколько вопросов — так ваш врач лучше поймёт, как вы сейчас."

config/texts/veteran_onboarding.yaml

@@ -0,0 +1,92 @@
+# Тексты стартового онбординга ветерана (issue #36): вступление, вопрос о травматическом опыте, пол для AUDIT-C,
+# вопрос о доступе к оружию, блок безопасного хранения, итог, FAQ.
+# СТАТУС: черновик, утверждает психиатр (#8). Ответы FAQ — формулировку проверяет юрист.
+#
+# Правила (проверяются тестом tests/test_dialog_veteran_onboarding.py):
+#   * тон — как у сослуживца: на «вы», коротко, без жалости и без нравоучений;
+#   * без слов «диагноз», «расстройство», «психолог», «психиатр», «болезнь», «псих»;
+#   * о сервисе — никогда «учёт», «контроль», «надзор» (страх пациента можно процитировать только в подписи
+#     кнопки FAQ — так человек узнаёт свой вопрос);
+#   * пациент не видит баллов и названий шкал.
+
+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]
+
+intro:
+  text: |-
+    Рад, что вы здесь. Пара слов, прежде чем начнём.
+
+    После возвращения многие замечают одно и то же: сон сбит, в голову лезет прошлое, тело всё ещё «на дежурстве», раздражает то, что раньше не трогало, тянет выпить, чтобы отпустило. Это частая реакция на то, через что вы прошли, а не слабость — и с этим реально работают.
+
+    Сейчас будет несколько коротких вопросов про сон, напряжение, настроение и алкоголь — минут пять, почти всё кнопками. Ответы увидит только ваш лечащий врач: так он сможет помогать точнее.
+  go: "Начнём"
+  repeat: "Когда будете готовы — нажмите «Начнём». Если есть вопросы, ответы на частые — кнопками ниже."
+  resume: "Давайте продолжим с того места, где остановились. Осталось немного."
+
+buttons_hint: "Пожалуйста, выберите вариант кнопкой ниже."
+
+exposure:
+  question: |-
+    Бывало ли с вами событие, которое сильно напугало, потрясло или было по-настоящему опасным для жизни — на службе, в бою или в другое время?
+    Подробности рассказывать не нужно.
+  answer_yes: "Да"
+  answer_no: "Нет"
+  decline: "Не хочу отвечать"
+  ack_yes: "Понял. Несколько вопросов о том, как это даёт о себе знать сейчас."
+  ack_other: "Хорошо, идём дальше."
+
+sex:
+  question: "Один уточняющий вопрос: ваш пол? Он нужен только для того, чтобы правильно учесть ответы про алкоголь — нормы для мужчин и женщин разные."
+  male: "Мужской"
+  female: "Женский"
+  decline: "Не хочу отвечать"
+
+weapon:
+  question: |-
+    Последний вопрос, он про безопасность, а не про проверку: есть ли у вас сейчас доступ к оружию — табельному, охотничьему, наградному или любому другому?
+  no_access: "Нет"
+  self: "Есть, хранится у меня"
+  others_or_safe: "Есть, у других / в сейфе"
+  decline: "Не хочу отвечать"
+  ack: "Спасибо, что ответили."
+  # Блок безопасного хранения — фиксированный текст, показывается только при доступе к оружию И сигнале риска.
+  safety_block_self: |-
+    Одна просьба — без нравоучений. Пока вам тяжело, пусть оружие и патроны побудут у того, кому вы доверяете: у брата, друга, в сейфе, ключ от которого у другого человека. Дело не в недоверии к вам. В трудную минуту важно, чтобы между мыслью и действием было время — это одна из самых надёжных мер защиты, и так делают многие.
+    Это можно спокойно обсудить с врачом на приёме.
+  safety_block_stored: |-
+    Хорошо, что оружие хранится не под рукой. Пусть пока так и остаётся: ключ или код — у человека, которому вы доверяете, а патроны — отдельно. В трудную минуту важно, чтобы между мыслью и действием было время.
+    Это можно спокойно обсудить с врачом на приёме.
+  safety_block_unknown: |-
+    Понимаю, что не хочется об этом говорить. Просто на всякий случай: если дома есть оружие или большой запас лекарств, пусть пока они побудут у человека, которому вы доверяете. В трудную минуту важно, чтобы между мыслью и действием было время.
+
+result:
+  intro: "Спасибо, что ответили на всё. Коротко, что видно по ответам:"
+  calm: "Сейчас серьёзных трудностей по ответам не видно. Хорошо — будем держать руку на пульсе."
+  load: "Нагрузка сейчас заметная. Так бывает у многих после возвращения, и это то, с чем врач может помочь."
+  ptsd: "• Прошлое пока не отпускает: напряжение, воспоминания, сон."
+  mood: "• Настроение и силы ниже, чем хотелось бы."
+  alcohol: "• Алкоголь — стоит обсудить с врачом, особенно если назначат препараты: они с алкоголем плохо сочетаются."
+  next: "Ваши ответы уже у лечащего врача — он обсудит их с вами на приёме. А я буду на связи каждый день с одним коротким вопросом и напоминанием о препарате."
+
+faq:
+  who:
+    button: "Кто увидит мои ответы?"
+    # формулировку проверяет юрист
+    text: |-
+      Только ваш лечащий врач — тот, кто дал вам код подключения. Перед автоматической обработкой из сообщений удаляются имена, телефоны и другие личные данные. Данные хранятся в России.
+      Отозвать согласие и остановить обработку можно в любой момент — кнопкой «Отозвать согласие» или командой /revoke.
+  unit:
+    button: "Это попадёт на учёт / в часть?"
+    # формулировку проверяет юрист
+    text: |-
+      Нет. Это добровольная консультативная помощь между визитами к врачу. Ваши ответы не передаются в часть, военкомат, работодателю или кому-то ещё без вашего согласия — кроме случаев, прямо предусмотренных законом. Отказ от помощника никак не влияет на ваше лечение.
+  licences:
+    button: "Заберут права на оружие или вождение?"
+    # формулировку проверяет юрист
+    text: |-
+      Помощник не выдаёт заключений и не отправляет ваши ответы ни в ГИБДД, ни в Росгвардию, ни на медкомиссии. Допуск к оружию и вождению решает отдельная медицинская комиссия по своим правилам, установленным законом.
+      Как именно это связано с вашим лечением — спросите лечащего врача прямо. Это нормальный вопрос, его задают многие, и честный ответ лучше, чем откладывать помощь из-за неизвестности.

tests/test_contracts.py

@@ -1,14 +1,24 @@
 from fastapi.testclient import TestClient
 
+from app.db import session
+from app.db.bootstrap import init_db
 from app.dialog.contracts import Event, get_engine
 from app.main import app
 from app.services.contracts import get_monitoring
 
+MEM = "sqlite+aiosqlite:///:memory:"
+
 
 async def test_stubs_are_usable() -> None:
-    replies = await get_engine().handle("tok", Event(type="text", text="привет"))
+    await session.configure(identity_url=MEM, clinical_url=MEM)
-    assert replies and replies[0].text
+    await init_db()
-    assert await get_monitoring().after_data_change("tok") == []
+    try:
+        replies = await get_engine().handle("tok", Event(type="text", text="привет"))
+        assert replies and replies[0].text
+        assert await get_monitoring().after_data_change("tok") == []  # unknown patient → no actions
+        assert await get_monitoring().is_in_crisis("tok") is False
+    finally:
+        await session.configure()
 
 
 def test_api_fixture_routes() -> None:

tests/test_db.py

@@ -70,6 +70,7 @@ async def test_create_all_puts_tables_in_the_right_db(dbs: None) -> None:
         "questionnaire_records",
         "questionnaire_notices",
         "questionnaire_tasks",
+        "scheduler_events",
     } == clinical_tables
     await init_db()  # idempotent
 

tests/test_dialog_engine_gaps.py

@@ -0,0 +1,385 @@
+"""Engine gaps found by the 13-day run (#56): no same-day C-SSRS repeat, scale intro registry, prefix triggers
+with a button payload (medication buttons carry the dose), abandoned-step timeout."""
+
+from collections.abc import AsyncIterator
+from datetime import datetime, timedelta
+from typing import Any
+
+import pytest
+from sqlalchemy import select
+
+from app.config import Settings
+from app.core import clock
+from app.core.rules import Action
+from app.core.scales import load_scale
+from app.db import repo, session
+from app.db.bootstrap import init_db
+from app.db.clinical import Dose, ItemResponse, Prescription
+from app.dialog import engine as engine_module
+from app.dialog import fsm, scale_session
+from app.dialog.contracts import Event, Reply
+from app.dialog.engine import STEP_AT, Engine
+from app.dialog.flows.meds import load_meds_texts
+from app.dialog.flows.veteran_daily import load_daily_texts
+from app.dialog.fsm import Ctx, Step, get_trigger, is_registered, trigger, trigger_payload
+from app.dialog.texts import load_dialog_texts
+from app.llm.gateway import LLMGateway
+from app.llm.mock import MockTasks
+
+TOKEN = "g" * 32
+OTHER = "h" * 32
+MEM = "sqlite+aiosqlite:///:memory:"
+NOW = datetime(2026, 10, 7, 9, 0)
+TEXTS = load_dialog_texts()
+MEDS = load_meds_texts()
+DAILY = load_daily_texts()
+SCREEN = TEXTS.settings.risk_screen_scale
+
+
+class ScriptedMonitoring:
+    def __init__(self) -> None:
+        self.actions: list[Action] = []
+
+    async def after_data_change(self, token: str) -> list[Action]:
+        out, self.actions = self.actions, []
+        return out
+
+    async def is_in_crisis(self, token: str) -> bool:
+        return True
+
+
+@pytest.fixture
+async def mon(monkeypatch: pytest.MonkeyPatch) -> AsyncIterator[ScriptedMonitoring]:
+    monkeypatch.setattr(clock, "get_settings", lambda: Settings(_env_file=None, demo_clock=True))
+    clock.set_now(NOW)
+    await session.configure(identity_url=MEM, clinical_url=MEM)
+    await init_db()
+    async with session.get_clinical_session() as s:
+        await repo.upsert_profile(
+            s, token=TOKEN, doctor_id="doc", modules=["veteran"], sex="m", birth_year=1990, therapy_start=NOW.date()
+        )
+        s.add(Prescription(token=TOKEN, drug="сертралин", dose="50 мг", schedule={"times": ["08:00"]}))
+        await repo.set_dialog_state(s, token=TOKEN, state="idle", data={"consent": True}, now=clock.now())
+    yield ScriptedMonitoring()
+    clock.reset()
+    await session.configure()
+
+
+@pytest.fixture
+def engine(mon: ScriptedMonitoring) -> Engine:
+    return Engine(gateway=LLMGateway(MockTasks()), monitoring=mon)
+
+
+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))
+
+
+async def system(e: Engine, kind: str, **params: Any) -> list[Reply]:
+    return await e.handle(TOKEN, Event(type="system", system=kind, params=params))  # type: ignore[arg-type]
+
+
+async def answer(e: Engine, scale_id: str, values: list[int]) -> list[Reply]:
+    out: list[Reply] = []
+    for item, value in zip(load_scale(scale_id).items, values, strict=False):
+        out += await press(e, f"scale:{scale_id}:{item.id}:{value}")
+    return out
+
+
+def buttons(replies: list[Reply]) -> list[str]:
+    return [b.data for r in replies for row in r.buttons for b in row]
+
+
+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(scale: str) -> list[Any]:
+    async with session.get_clinical_session() as s:
+        return list(await repo.list_scale_results(s, TOKEN, scale))
+
+
+async def checkin() -> Any:
+    async with session.get_clinical_session() as s:
+        rows = await repo.list_checkins(s, TOKEN)
+        return rows[-1] if rows else None
+
+
+async def doses() -> list[Dose]:
+    async with session.get_clinical_session() as s:
+        return list((await s.scalars(select(Dose).where(Dose.token == TOKEN).order_by(Dose.id))).all())
+
+
+# ----------------------------------------------------------------------------------------------- 1. C-SSRS repeat
+
+
+async def test_item9_right_after_cssrs_does_not_repeat_screen(engine: Engine, mon: ScriptedMonitoring) -> None:
+    await system(engine, "scale_due", scale=SCREEN, scheduled=False)
+    assert (await get_state())[0] == f"scale.{SCREEN}"
+    await answer(engine, SCREEN, [0] * 6)
+    clock.advance(hours=1)
+    await system(engine, "scale_due", scale="phq9", scheduled=False)
+    # The rule phq9_item9 also asks for the screen — neither path may repeat it within the cooldown.
+    mon.actions = [Action(type="schedule_scale", code="phq9_item9", params={"scale": SCREEN, "scheduled": False})]
+    r = await answer(engine, "phq9", [1] * 8 + [1])
+    assert any(x.kind == "crisis" for x in r)  # the fixed crisis script still goes out
+    state, data = await get_state()
+    assert state == "crisis" and data["crisis"]["open"] is True and not data.get("pending_scales")
+    assert len(await results(SCREEN)) == 1
+
+
+async def test_item9_after_old_cssrs_asks_screen_again(engine: Engine) -> None:
+    await system(engine, "scale_due", scale=SCREEN, scheduled=False)
+    await answer(engine, SCREEN, [0] * 6)
+    clock.advance(hours=TEXTS.settings.cssrs_cooldown_hours + 1)
+    await system(engine, "scale_due", scale="phq9", scheduled=False)
+    r = await answer(engine, "phq9", [1] * 8 + [1])
+    assert any(x.kind == "crisis" for x in r)
+    assert (await get_state())[0] == f"scale.{SCREEN}" and TEXTS.scale.risk_screen_intro in r[-1].text
+
+
+async def test_unfinished_cssrs_does_not_count(engine: Engine) -> None:
+    await system(engine, "scale_due", scale=SCREEN, scheduled=False)
+    await answer(engine, SCREEN, [0, 0])  # left halfway: no complete result
+    await system(engine, "scale_due", scale="phq9", scheduled=False)  # queued behind the screen
+    async with session.get_clinical_session() as s:
+        assert not await scale_session.recent_risk_screen(_ctx(s), SCREEN)
+
+
+def _ctx(s: Any, state: str = "idle", data: dict[str, Any] | None = None) -> Ctx:
+    return Ctx(
+        token=TOKEN,
+        state=state,
+        data=data or {},
+        now=clock.now(),
+        session=s,
+        gateway=LLMGateway(MockTasks()),
+        monitoring=ScriptedMonitoring(),
+        texts=TEXTS,
+    )
+
+
+# ----------------------------------------------------------------------------------------------- 2. intros
+
+
+async def test_registered_intro_text_and_callable(engine: Engine, monkeypatch: pytest.MonkeyPatch) -> None:
+    seen: list[tuple[str, bool, str]] = []
+
+    def intro(ctx: Ctx, scale_id: str, scheduled: bool, return_state: str) -> str | None:
+        seen.append((scale_id, scheduled, return_state))
+        return "Вступление к GAD-7"
+
+    monkeypatch.setattr(scale_session, "_intros", dict(scale_session._intros))
+    scale_session.register_scale_intro("gad7", intro)
+    with pytest.raises(ValueError):
+        scale_session.register_scale_intro("gad7", "второе")
+    r = await system(engine, "scale_due", scale="gad7", scheduled=True)
+    assert r[0].text.startswith("Вступление к GAD-7") and seen == [("gad7", True, "idle")]
+
+    scale_session.register_scale_intro("gad7", "Текстовое вступление", replace=True)
+    async with session.get_clinical_session() as s:
+        step = await scale_session.start_scale(_ctx(s), "gad7", return_state="idle")
+        assert step.data["scales"]["gad7"]["intro"] == "Текстовое вступление"
+        explicit = await scale_session.start_scale(_ctx(s), "gad7", return_state="idle", intro="Своё")
+        assert explicit.data["scales"]["gad7"]["intro"] == "Своё"
+
+
+def test_daily_intros_use_registry_not_enter_hooks() -> None:
+    for scale_id in DAILY.settings.intro_scales:
+        assert not is_registered(f"scale.{scale_id}")  # no per-scale enter hooks any more
+        assert scale_id in scale_session._intros
+
+
+async def test_unscheduled_intro_from_registry(engine: Engine) -> None:
+    r = await system(engine, "scale_due", scale="phq9", scheduled=False)
+    assert DAILY.scale_intros.unscheduled["phq9"] in r[0].text
+
+
+# ----------------------------------------------------------------------------------------------- 3. payload buttons
+
+
+async def test_prefix_trigger_payload_and_precedence(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(fsm, "_triggers", dict(fsm._triggers))
+    monkeypatch.setattr(fsm, "_prefix_triggers", dict(fsm._prefix_triggers))
+
+    def handler(tag: str) -> Any:
+        async def fn(ctx: Ctx, event: Event) -> Step:
+            return Step(data={"tag": tag, "payload": trigger_payload(event)})
+
+        return fn
+
+    trigger("button:zz:a:*")(handler("short"))
+    trigger("button:zz:a:b:*")(handler("long"))
+    trigger("button:zz:a:exact")(handler("exact"))
+    with pytest.raises(ValueError):
+        trigger("button:zz:a:*")(handler("dup"))
+
+    async def run(key: str) -> dict[str, Any] | None:
+        fn = get_trigger(key)
+        if fn is None:
+            return None
+        return (await fn(None, Event(type="button", button=key.split(":", 1)[1]))).data  # type: ignore[arg-type]
+
+    assert await run("button:zz:a:42") == {"tag": "short", "payload": "42"}
+    assert await run("button:zz:a:b:7:x") == {"tag": "long", "payload": "7:x"}
+    assert await run("button:zz:a:exact") == {"tag": "exact", "payload": None}
+    assert await run("button:zz:a:") is None and await run("button:zz:a") is None
+
+
+async def test_reminder_buttons_carry_dose_and_no_state_key(engine: Engine) -> None:
+    async with session.get_clinical_session() as s:
+        await repo.set_dialog_state(
+            s, token=TOKEN, state="idle", data={"consent": True, "med_pending": {"dose_id": 1}}, now=clock.now()
+        )
+    r = await system(engine, "med_reminder")
+    [dose] = await doses()
+    assert buttons(r) == [f"med:taken:{dose.id}", f"med:skip:{dose.id}", f"med:side_effects:{dose.id}"]
+    assert "med_pending" not in (await get_state())[1]
+    skip = await press(engine, f"med:skip:{dose.id}")
+    assert f"med:skip_reason:side_effects:{dose.id}" in buttons(skip)
+    se = await press(engine, f"med:skip_reason:side_effects:{dose.id}")
+    await press(engine, f"med:se:nausea:{dose.id}")
+    [dose] = await doses()
+    assert dose.status == "skipped" and dose.side_effects == "nausea" and "med:se:nausea:" in buttons(se)[0]
+
+
+async def test_two_reminders_each_button_answers_its_own_dose(engine: Engine) -> None:
+    first = await system(engine, "med_reminder")
+    second = await system(engine, "med_reminder")
+    await press(engine, buttons(first)[0])  # «Принял» of the FIRST reminder, pressed after the second arrived
+    d1, d2 = await doses()
+    assert (d1.status, d2.status) == ("taken", "pending")
+    await press(engine, buttons(second)[1])
+    assert [d.status for d in await doses()] == ["taken", "skipped"]
+
+
+@pytest.mark.parametrize("data", ["med:taken", "med:skip", "med:se:nausea", "med:dose:yes", "med:skip_reason:quit"])
+async def test_legacy_payloadless_buttons_are_stale(engine: Engine, data: str) -> None:
+    await system(engine, "med_reminder")
+    r = await press(engine, data)
+    assert r[0].text == MEDS.stale and await checkin() is None
+    assert [d.status for d in await doses()] == ["pending"]
+
+
+async def test_foreign_unknown_and_old_dose_buttons_are_stale(engine: Engine) -> None:
+    async with session.get_clinical_session() as s:
+        rx = Prescription(token=OTHER, drug="сертралин", dose="50 мг", schedule={"times": ["08:00"]})
+        s.add(rx)
+        await s.flush()
+        foreign = Dose(token=OTHER, prescription_id=rx.id, planned_at=clock.now(), status="pending")
+        s.add(foreign)
+        await s.flush()
+        foreign_id = foreign.id
+    for data in (f"med:taken:{foreign_id}", "med:taken:9999", "med:taken:garbage", "med:taken:2026-10-06"):
+        assert (await press(engine, data))[0].text == MEDS.stale
+    old = await system(engine, "med_reminder")
+    clock.advance(days=1)
+    assert (await press(engine, buttons(old)[0]))[0].text == MEDS.stale
+    assert await checkin() is None
+
+
+async def test_reminder_without_dose_row_uses_day_payload(engine: Engine) -> None:
+    async with session.get_clinical_session() as s:
+        for row in await s.scalars(select(Prescription)):
+            await s.delete(row)
+    r = await system(engine, "med_reminder")
+    assert buttons(r)[0] == f"med:taken:{NOW.date().isoformat()}"
+    r = await press(engine, buttons(r)[0])
+    assert r[0].text == MEDS.taken_ack and (await checkin()).med_taken is True and await doses() == []
+
+
+async def test_checkin_med_step_answers_todays_dose(engine: Engine) -> None:
+    await system(engine, "med_reminder")
+    await system(engine, "daily_checkin")
+    await press(engine, "daily:mood:5")
+    await press(engine, "daily:med:yes")
+    assert [d.status for d in await doses()] == ["taken"]
+
+
+# ----------------------------------------------------------------------------------------------- 4. abandon timeout
+
+
+async def test_abandoned_checkin_returns_to_idle(engine: Engine) -> None:
+    await system(engine, "daily_checkin")
+    await press(engine, "daily:mood:3")
+    assert (await get_state())[0].startswith("daily.")
+    clock.advance(hours=TEXTS.settings.abandon_after_hours + 1)
+    r = await system(engine, "med_reminder")
+    state, data = await get_state()
+    assert state == "idle" and "checkin" not in data
+    assert len(r) == 1 and r[0].text.startswith(MEDS.reminder.text.split("{")[0])  # no check-in question repeated
+    assert (await checkin()).mood == 3  # stored answer survives
+
+
+async def test_short_pause_keeps_checkin(engine: Engine) -> None:
+    await system(engine, "daily_checkin")
+    clock.advance(hours=TEXTS.settings.abandon_after_hours - 1)
+    r = await press(engine, "daily:mood:6")
+    assert (await get_state())[0] != "idle" and r
+
+
+async def test_abandoned_scale_lets_queued_scale_start(engine: Engine) -> None:
+    await system(engine, "scale_due", scale="phq9", scheduled=True)
+    await answer(engine, "phq9", [1, 2])
+    clock.advance(hours=4)
+    await system(engine, "scale_due", scale="pcl5", scheduled=True)  # queued; system events don't extend the step
+    clock.advance(hours=3)
+    r = await text(engine, "добрый вечер")
+    state, data = await get_state()
+    assert state == "scale.pcl5" and "phq9" not in data["scales"] and not data.get("pending_scales")
+    assert DAILY.scale_intros.scheduled["pcl5"] in r[-1].text
+    assert await results("phq9") == []  # unfinished: no total, no severity
+    async with session.get_clinical_session() as s:
+        stored = (await s.scalars(select(ItemResponse).where(ItemResponse.scale == "phq9"))).all()
+    assert [x.score for x in stored] == [1, 2]
+
+
+async def test_timeout_never_applies_in_crisis(engine: Engine) -> None:
+    await system(engine, "scale_due", scale="phq9", scheduled=False)
+    await answer(engine, "phq9", [0] * 8 + [2])  # item 9 → crisis + C-SSRS (none done before)
+    state, data = await get_state()
+    assert state == f"scale.{SCREEN}" and data["crisis"]["open"] is True
+    clock.advance(hours=TEXTS.settings.abandon_after_hours * 3)
+    await press(engine, f"scale:{SCREEN}:cssrs_1:0")
+    state, data = await get_state()
+    assert state == f"scale.{SCREEN}" and data["scales"][SCREEN]["idx"] == 1
+
+
+async def test_timeout_never_skips_onboarding_scale(engine: Engine) -> None:
+    item = load_scale("pcl5").items[0]
+    old = (clock.now() - timedelta(hours=48)).isoformat()
+    session_data = {
+        "idx": 0,
+        "answers": {},
+        "return_state": "veteran.onboarding.phq9",
+        "scheduled": True,
+        "intro": None,
+        "intro_shown": True,
+    }
+    async with session.get_clinical_session() as s:
+        await repo.set_dialog_state(
+            s,
+            token=TOKEN,
+            state="scale.pcl5",
+            data={"consent": True, "scales": {"pcl5": session_data}, STEP_AT: old},
+            now=clock.now(),
+        )
+    await press(engine, f"scale:pcl5:{item.id}:1")
+    state, data = await get_state()
+    assert state == "scale.pcl5" and data["scales"]["pcl5"]["idx"] == 1
+
+
+async def test_timeout_disabled_same_day(engine: Engine, monkeypatch: pytest.MonkeyPatch) -> None:
+    clock.set_now(NOW.replace(hour=1))
+    off = TEXTS.model_copy(update={"settings": TEXTS.settings.model_copy(update={"abandon_after_hours": 0})})
+    monkeypatch.setattr(engine_module, "load_dialog_texts", lambda: off)
+    await system(engine, "daily_checkin")
+    clock.advance(hours=20)
+    await press(engine, "daily:mood:4")
+    assert (await get_state())[0] != "idle" and (await checkin()).mood == 4

tests/test_dialog_meds.py

@@ -0,0 +1,377 @@
+"""Medication flow and check-in units (#37): reminder buttons, skip reasons, side effects, medication question,
+med_info rule reactions, check-in edge cases, text policy of the YAML files."""
+
+from collections.abc import AsyncIterator
+from datetime import datetime, timedelta
+from pathlib import Path
+from typing import Any
+
+import pytest
+import yaml
+from sqlalchemy import select
+
+from app.config import Settings, get_settings
+from app.core import clock
+from app.core.rules import Action
+from app.db import repo, session
+from app.db.bootstrap import init_db
+from app.db.clinical import Dose, Prescription
+from app.dialog.contracts import Event, Reply
+from app.dialog.engine import Engine
+from app.dialog.flows.meds import load_meds_texts
+from app.dialog.flows.veteran_daily import load_daily_texts
+from app.dialog.fsm import Ctx, enter_hook_for
+from app.dialog.scale_session import start_scale
+from app.dialog.texts import load_dialog_texts
+from app.integrations.base import DrugSection, integrations
+from app.llm import med_qa as med_qa_module
+from app.llm.gateway import LLMGateway
+from app.llm.med_qa import fallback_text
+from app.llm.mock import MockTasks
+
+TOKEN = "m" * 32
+MEM = "sqlite+aiosqlite:///:memory:"
+NOW = datetime(2026, 10, 7, 8, 0)
+MEDS = load_meds_texts()
+DAILY = load_daily_texts()
+SECTIONS = {
+    "side_effects": "Тошнота обычно возникает в первые дни приёма и проходит в течение 1–2 недель.",
+    "onset": "Улучшение обычно отмечается через 2–4 недели регулярного приёма.",
+    "alcohol": "Совместный приём с алкоголем не рекомендуется.",
+    "administration": "Препарат принимают один раз в сутки, утром или вечером, независимо от еды.",
+}
+
+
+class FakeDrugInfo:
+    async def sections(self, inn: str) -> list[DrugSection]:
+        return [DrugSection(inn=inn, section=k, text=v, source="синтетика") for k, v in SECTIONS.items()]
+
+
+class ScriptedMonitoring:
+    def __init__(self) -> None:
+        self.calls = 0
+        self.actions: list[Action] = []
+
+    async def after_data_change(self, token: str) -> list[Action]:
+        self.calls += 1
+        out, self.actions = self.actions, []
+        return out
+
+    async def is_in_crisis(self, token: str) -> bool:
+        return False
+
+
+@pytest.fixture
+async def mon(monkeypatch: pytest.MonkeyPatch) -> AsyncIterator[ScriptedMonitoring]:
+    monkeypatch.setattr(clock, "get_settings", lambda: Settings(_env_file=None, demo_clock=True))
+    monkeypatch.setattr(integrations, "drug_info", FakeDrugInfo())
+    monkeypatch.setattr(med_qa_module, "get_gateway", lambda: LLMGateway(MockTasks()))
+    clock.set_now(NOW)
+    await session.configure(identity_url=MEM, clinical_url=MEM)
+    await init_db()
+    async with session.get_clinical_session() as s:
+        await repo.upsert_profile(
+            s, token=TOKEN, doctor_id="doc", modules=["veteran"], sex="m", birth_year=1990, therapy_start=NOW.date()
+        )
+        s.add(Prescription(token=TOKEN, drug="сертралин", dose="50 мг", schedule={"times": ["08:00"]}))
+        await repo.set_dialog_state(s, token=TOKEN, state="idle", data={"consent": True}, now=clock.now())
+    yield ScriptedMonitoring()
+    clock.reset()
+    await session.configure()
+
+
+@pytest.fixture
+def engine(mon: ScriptedMonitoring) -> Engine:
+    return Engine(gateway=LLMGateway(MockTasks()), monitoring=mon)
+
+
+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))
+
+
+async def system(e: Engine, kind: str, **params: Any) -> list[Reply]:
+    return await e.handle(TOKEN, Event(type="system", system=kind, params=params))  # type: ignore[arg-type]
+
+
+def buttons(replies: list[Reply]) -> list[str]:
+    return [b.data for r in replies for row in r.buttons for b in row]
+
+
+def kinds(replies: list[Reply]) -> list[str]:
+    """Button data without the dose payload (#56): "med:taken:7" → "med:taken"."""
+    return [d.rsplit(":", 1)[0] if d.startswith("med:") else d for d in buttons(replies)]
+
+
+async def tap(e: Engine, replies: list[Reply], kind: str) -> list[Reply]:
+    """Press the button of `replies` whose data (without payload) is `kind` — as the patient would."""
+    [data] = [d for d in buttons(replies) if d.startswith("med:") and d.rsplit(":", 1)[0] == kind]
+    return await press(e, data)
+
+
+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 checkin() -> Any:
+    async with session.get_clinical_session() as s:
+        rows = await repo.list_checkins(s, TOKEN)
+        return rows[-1] if rows else None
+
+
+async def doses() -> list[Dose]:
+    async with session.get_clinical_session() as s:
+        return list((await s.scalars(select(Dose).where(Dose.token == TOKEN).order_by(Dose.id))).all())
+
+
+# ----------------------------------------------------------------------------------------------- reminder
+
+
+async def test_reminder_without_dose_id_creates_dose_and_taken(engine: Engine, mon: ScriptedMonitoring) -> None:
+    r = await system(engine, "med_reminder")
+    assert r[0].text == MEDS.reminder.text.format(drug="сертралин", dose="50 мг")
+    [dose] = await doses()
+    assert buttons(r) == [f"med:taken:{dose.id}", f"med:skip:{dose.id}", f"med:side_effects:{dose.id}"]
+    assert dose.status == "pending"
+    r = await tap(engine, r, "med:taken")
+    assert r[0].text == MEDS.taken_ack and mon.calls == 1
+    [dose] = await doses()
+    assert dose.status == "taken" and dose.answered_at is not None
+    assert (await checkin()).med_taken is True
+
+
+async def test_stale_reminder_button(engine: Engine) -> None:
+    reminder = await system(engine, "med_reminder")
+    clock.advance(days=1)
+    r = await tap(engine, reminder, "med:taken")
+    assert r[0].text == MEDS.stale and await checkin() is None
+
+
+@pytest.mark.parametrize("reason", ["forgot", "other"])
+async def test_skip_simple_reasons(engine: Engine, reason: str) -> None:
+    r = await tap(engine, await system(engine, "med_reminder"), "med:skip")
+    assert kinds(r) == [f"med:skip_reason:{x}" for x in ("forgot", "side_effects", "quit", "other")]
+    r = await tap(engine, r, f"med:skip_reason:{reason}")
+    assert r[0].text == MEDS.skip.ack[reason]
+    c = await checkin()
+    assert c.med_taken is False and not c.quit_intent
+    assert [d.status for d in await doses()] == ["skipped"]
+
+
+async def test_skip_quit_sets_intent_and_explains(engine: Engine, mon: ScriptedMonitoring) -> None:
+    skip = await tap(engine, await system(engine, "med_reminder"), "med:skip")
+    mon.actions = [Action(type="med_info", code="quit_intent", params={"topic": "delayed_effect"})]
+    r = await tap(engine, skip, "med:skip_reason:quit")
+    assert r[0].text == MEDS.skip.ack["quit"]
+    assert "2–4 недели" in r[1].text and MEDS.med_info["delayed_effect"].notice in r[1].text
+    assert (await checkin()).quit_intent is True
+
+
+async def test_skip_because_of_side_effects_asks_which(engine: Engine) -> None:
+    skip = await tap(engine, await system(engine, "med_reminder"), "med:skip")
+    r = await tap(engine, skip, "med:skip_reason:side_effects")
+    assert "med:se:nausea" in kinds(r) and "med:se:other" in kinds(r)
+
+
+async def test_side_effect_button(engine: Engine, mon: ScriptedMonitoring) -> None:
+    r = await tap(engine, await system(engine, "med_reminder"), "med:side_effects")
+    assert kinds(r) == [f"med:se:{c}" for c in ("nausea", "insomnia", "headache", "dizziness", "agitation", "other")]
+    r = await tap(engine, r, "med:se:nausea")
+    assert "проходит в течение 1–2 недель" in r[0].text  # instruction excerpt
+    assert r[1].text == MEDS.side_effects.ask_dose and kinds(r[1:]) == ["med:dose:yes", "med:dose:no"]
+    assert (await checkin()).side_effects == ["nausea"]
+    assert (await doses())[0].side_effects == "nausea"
+    r = await tap(engine, r, "med:dose:yes")
+    assert (await checkin()).med_taken is True and (await doses())[0].status == "taken"
+
+
+async def test_side_effect_other_free_text(engine: Engine) -> None:
+    reminder = await system(engine, "med_reminder")
+    r = await tap(engine, await tap(engine, reminder, "med:side_effects"), "med:se:other")
+    assert r[0].text == MEDS.side_effects.describe and (await get_state())[0] == "meds.side_effect_text"
+    r = await text(engine, "голова болит весь день")
+    assert (await checkin()).side_effects == ["headache"] and (await get_state())[0] == "idle"
+    await tap(engine, await tap(engine, reminder, "med:side_effects"), "med:se:other")
+    await text(engine, "какое-то странное ощущение")
+    assert (await checkin()).side_effects == ["headache", "other"]
+
+
+async def test_same_side_effect_as_yesterday_gets_support(engine: Engine) -> None:
+    async with session.get_clinical_session() as s:
+        await repo.upsert_checkin(
+            s, token=TOKEN, day=clock.today() - timedelta(days=1), now=clock.now(), side_effects=["nausea"]
+        )
+    se = await tap(engine, await system(engine, "med_reminder"), "med:side_effects")
+    r = await tap(engine, se, "med:se:nausea")
+    assert r[0].text == MEDS.side_effects.repeat
+
+
+# ----------------------------------------------------------------------------------------------- question menu
+
+
+async def test_med_question_menu(engine: Engine) -> None:
+    r = await press(engine, "menu:med_question")
+    assert r[0].text == MEDS.question.prompt and (await get_state())[0] == "meds.question"
+    r = await text(engine, "можно ли пить алкоголь?")
+    assert "Совместный приём с алкоголем не рекомендуется" in r[0].text and (await get_state())[0] == "idle"
+    await press(engine, "menu:med_question")
+    r = await text(engine, "можно увеличить дозу до 100 мг?")
+    assert r[0].text == fallback_text()
+
+
+async def test_med_question_menu_left_by_button(engine: Engine) -> None:
+    await press(engine, "menu:med_question")
+    r = await press(engine, "menu:checkin")
+    assert r[0].text == DAILY.checkin.start and (await get_state())[0] == "daily.mood"
+
+
+async def test_med_question_without_prescription(engine: Engine) -> None:
+    async with session.get_clinical_session() as s:
+        for row in await s.scalars(select(Prescription)):
+            await s.delete(row)
+    r = await press(engine, "menu:med_question")
+    assert r[0].text == MEDS.question.no_prescription and (await get_state())[0] == "idle"
+
+
+async def test_med_info_without_excerpt_uses_fixed_text(
+    engine: Engine, mon: ScriptedMonitoring, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    class Empty:
+        async def sections(self, inn: str) -> list[DrugSection]:
+            return []
+
+    monkeypatch.setattr(integrations, "drug_info", Empty())
+    mon.actions = [Action(type="med_info", code="alcohol_3d", params={"topic": "alcohol_interaction"})]
+    r = await press(engine, "menu:visit")  # any data-writing step runs monitoring
+    spec = MEDS.med_info["alcohol_interaction"]
+    assert r[0].text == DAILY.visit.requested
+    assert r[1].text == "\n\n".join([spec.lead, spec.fallback, spec.notice])
+    assert (await get_state())[1]["visit_requests"]
+
+
+async def test_idle_free_text_med_question_and_side_effect(engine: Engine, mon: ScriptedMonitoring) -> None:
+    r = await text(engine, "когда подействуют таблетки?")
+    assert "2–4 недели" in r[0].text and "menu:checkin" in buttons(r)
+    r = await text(engine, "подташнивает после таблеток")
+    assert "проходит в течение 1–2 недель" in r[0].text and (await checkin()).side_effects == ["nausea"]
+    assert mon.calls == 1
+
+
+# ----------------------------------------------------------------------------------------------- check-in edges
+
+
+async def test_checkin_asks_dose_when_not_answered_and_typed_mood(engine: Engine) -> None:
+    await system(engine, "daily_checkin")
+    r = await text(engine, "отлично")
+    assert DAILY.checkin.buttons_hint in r[0].text and (await get_state())[0] == "daily.mood"
+    r = await text(engine, "7")
+    assert r[0].text == MEDS.checkin.question.format(drug="сертралин")
+    await press(engine, "daily:med:no")
+    c = await checkin()
+    assert (c.mood, c.med_taken) == (7, False)
+    assert (await get_state())[0] == "daily.sleep"
+
+
+async def test_reminder_during_checkin_keeps_checkin(engine: Engine) -> None:
+    await system(engine, "daily_checkin")
+    r = await system(engine, "med_reminder")
+    assert kinds(r[:1]) == ["med:taken", "med:skip", "med:side_effects"] and r[-1].text == DAILY.checkin.mood.question
+    r = await tap(engine, r, "med:taken")
+    assert r[0].text == MEDS.taken_ack and (await get_state())[0] == "daily.mood"
+    await press(engine, "daily:mood:5")
+    assert (await get_state())[0] == "daily.sleep"  # dose already answered → no med question
+
+
+async def test_unfinished_checkin_dropped_next_day(engine: Engine) -> None:
+    await system(engine, "daily_checkin")
+    await press(engine, "daily:mood:2")
+    clock.advance(days=1)
+    r = await system(engine, "daily_checkin")
+    assert r[0].text == DAILY.checkin.start and (await get_state())[0] == "daily.mood"
+
+
+async def test_checkin_once_a_day(engine: Engine) -> None:
+    await system(engine, "daily_checkin")
+    for b in ("daily:mood:6", "daily:med:yes", "daily:sleep:6-8", "daily:nightmares:no", "daily:alcohol:none"):
+        await press(engine, b)
+    await press(engine, "daily:text:skip")
+    assert await system(engine, "daily_checkin") == []
+    r = await press(engine, "menu:checkin")
+    assert r[0].text == DAILY.checkin.already_done
+
+
+async def test_alcohol_clarify_for_non_veteran(engine: Engine) -> None:
+    async with session.get_clinical_session() as s:
+        await repo.upsert_profile(s, token=TOKEN, modules=["depression"])
+    await system(engine, "daily_checkin")
+    await press(engine, "daily:mood:5")
+    r = await press(engine, "daily:med:yes")
+    assert buttons(r) == ["daily:text:skip"]  # no sleep / alcohol questions outside the veteran module
+    r = await text(engine, "вечером выпил пива с друзьями")
+    assert r[-1].text == DAILY.checkin.alcohol.clarify and (await get_state())[0] == "daily.alcohol_amount"
+    r = await press(engine, "daily:alcohol_amount:more")
+    assert r[0].text == DAILY.checkin.done and (await checkin()).alcohol_bucket == "more"
+
+
+async def test_buttons_win_over_llm(engine: Engine) -> None:
+    await system(engine, "daily_checkin")
+    for b in ("daily:mood:5", "daily:med:yes", "daily:sleep:6-8", "daily:nightmares:no", "daily:alcohol:none"):
+        await press(engine, b)
+    await text(engine, "почти не спал, кошмары, брошу таблетки")
+    c = await checkin()
+    assert (c.sleep_bucket, c.nightmares, c.alcohol_bucket, c.quit_intent) == ("6-8", False, "none", True)
+    assert c.free_text_scrubbed
+
+
+async def test_scale_intro_not_added_to_onboarding(engine: Engine, mon: ScriptedMonitoring) -> None:
+    await system(engine, "scale_due", scale="dar5", scheduled=True)
+    state, data = await get_state()
+    assert state == "scale.dar5" and data["scales"]["dar5"]["intro"] == DAILY.scale_intros.scheduled["dar5"]
+
+    async with session.get_clinical_session() as s:
+        ctx = Ctx(
+            token=TOKEN,
+            state="idle",
+            data={},
+            now=clock.now(),
+            session=s,
+            gateway=LLMGateway(MockTasks()),
+            monitoring=mon,
+            texts=load_dialog_texts(),
+        )
+        step = await start_scale(ctx, "pcl5", return_state="veteran.onboarding.phq9")
+        ctx.data.update(step.data)
+        ctx.state = "scale.pcl5"
+        hook = enter_hook_for("scale.pcl5")
+        assert hook is not None
+        entered = await hook(ctx)
+    assert DAILY.scale_intros.scheduled["pcl5"] not in entered.replies[0].text
+
+
+# ----------------------------------------------------------------------------------------------- text policy
+
+
+def _strings(node: Any) -> list[str]:
+    if isinstance(node, str):
+        return [node]
+    if isinstance(node, dict):
+        return [s for v in node.values() for s in _strings(v)]
+    if isinstance(node, list):
+        return [s for v in node for s in _strings(v)]
+    return []
+
+
+@pytest.mark.parametrize("name", ["meds.yaml", "veteran_daily.yaml"])
+def test_texts_policy(name: str) -> None:
+    path: Path = get_settings().config_dir / "texts" / name
+    raw = path.read_text(encoding="utf-8")
+    assert "черновик, утверждает психиатр (#8)" in raw.splitlines()[2]
+    values = " ".join(_strings(yaml.safe_load(raw))).lower()
+    for word in ("учёт", "учет", "контрол", "надзор", "диагноз", "птср", "депресси", "увеличьте доз", "мг"):
+        assert word not in values, (name, word)

tests/test_dialog_veteran_days.py

@@ -0,0 +1,391 @@
+"""Acceptance: 13 days of the veteran Andrey (docs/product/01_demo_persona.md) through the real dialog engine.
+
+Andrey is set up post-onboarding directly in the DBs (no dependency on #36): modules veteran + depression,
+therapy day 0, sertraline 50 mg at 08:00, consent. Each day: the dose reminder (as the scheduler would send it)
+→ button; the daily check-in → buttons + the free text from the table.
+
+Lane B's MonitoringService (#38) is replaced by a double that builds a `PatientSnapshot` from the clinical DB,
+runs the REAL `app.core.rules.evaluate`, persists alert/crisis actions (so dedup works) and returns the dialog
+actions — the same split as the frozen contract (ADR-007). Drug instruction excerpts are synthetic (FakeDrugInfo).
+"""
+
+from collections.abc import AsyncIterator
+from datetime import date, datetime, timedelta
+from typing import Any
+
+import pytest
+from sqlalchemy import func, select
+
+from app.config import Settings
+from app.core import clock
+from app.core.rules import Action, DailyCheckin, DoseRecord, PatientSnapshot, ScalePoint, evaluate, patient_status
+from app.core.scales import load_scale
+from app.core.types import AlertLevel
+from app.db import identity_repo, repo, session
+from app.db.bootstrap import init_db
+from app.db.clinical import Dose, Message, Prescription
+from app.dialog.contracts import Event, Reply
+from app.dialog.engine import Engine
+from app.dialog.flows.meds import load_meds_texts
+from app.dialog.flows.veteran_daily import load_daily_texts
+from app.integrations.base import DrugSection, integrations
+from app.llm import med_qa as med_qa_module
+from app.llm.gateway import LLMGateway
+from app.llm.mock import MockTasks
+from app.services import contracts as monitoring_contracts
+
+TOKEN = "e" * 32
+MEM = "sqlite+aiosqlite:///:memory:"
+DAY0 = datetime(2026, 10, 5, 8, 0)  # visit day; reminders at 08:00, check-ins at 20:00 (Europe/Moscow)
+SRC = "ГРЛС: тестовая выдержка (синтетика)"
+SERTRALINE = {
+    "side_effects": (
+        "Наиболее частые нежелательные реакции в начале терапии: тошнота, диарея, сухость во рту. "
+        "Тошнота обычно возникает в первые дни приёма и проходит в течение 1–2 недель. "
+        "Для уменьшения тошноты препарат принимают после еды."
+    ),
+    "onset": (
+        "Терапевтический эффект развивается постепенно. "
+        "Улучшение обычно отмечается через 2–4 недели регулярного приёма. "
+        "Суточная доза может увеличиваться врачом до 200 мг."
+    ),
+    "alcohol": (
+        "Совместный приём с алкоголем не рекомендуется. "
+        "Алкоголь может ухудшать сон и настроение и усиливать побочные эффекты."
+    ),
+}
+
+# Persona table: day → (dose taken, mood, sleep, nightmares, alcohol, free text). "" = skip the free text.
+TABLE: dict[int, tuple[bool, int, str, bool, str, str]] = {
+    1: (True, 4, "4-6", True, "none", ""),
+    2: (True, 4, "4-6", True, "none", "подташнивает с утра"),
+    3: (True, 4, "4-6", True, "none", "опять тошнит"),
+    4: (True, 3, "4-6", False, "none", "тошнота не проходит"),
+    5: (False, 3, "4-6", False, "none", "от таблеток только хуже, брошу"),
+    6: (True, 4, "6-8", False, "none", "врач позвонил, продолжаю"),
+    7: (True, 4, "6-8", False, "none", ""),
+    8: (True, 4, "<4", True, "1-2", "снится одно и то же"),
+    9: (True, 4, "<4", True, "more", ""),
+    10: (True, 4, "<4", True, "more", "без пива не могу уснуть"),
+    11: (True, 3, "4-6", False, "1-2", ""),
+    12: (True, 3, "4-6", False, "none", ""),
+    13: (True, 2, "<4", True, "none", "всё бессмысленно, устал от всего"),
+}
+# Expected dashboard colour at the end of each day (persona: 🟢🟢🟢🟡🟡🟢🟢🟢🟢🟡🟡🟡🔴).
+COLORS = ["green"] * 3 + ["yellow"] * 2 + ["green"] * 4 + ["yellow"] * 3 + ["red"]
+
+
+class FakeDrugInfo:
+    async def sections(self, inn: str) -> list[DrugSection]:
+        if inn != "сертралин":
+            return []
+        return [DrugSection(inn=inn, section=k, text=v, source=SRC) for k, v in SERTRALINE.items()]
+
+
+class RuleMonitoring:
+    """Test double of lane B's MonitoringService: real rules over a snapshot of the clinical DB."""
+
+    def __init__(self) -> None:
+        self.fired: list[str] = []
+
+    async def _snapshot(self, s: Any, token: str) -> PatientSnapshot:
+        now = clock.now()
+        profile = await repo.get_profile(s, token)
+        assert profile is not None and profile.therapy_start is not None and profile.birth_year is not None
+        checkins = [
+            DailyCheckin(
+                day=c.day,
+                mood=c.mood,
+                med_taken=c.med_taken,
+                sleep_hours_bucket=c.sleep_bucket,  # type: ignore[arg-type]
+                alcohol=c.alcohol_bucket,  # type: ignore[arg-type]
+                side_effects=list(c.side_effects or []),
+                quit_intent=bool(c.quit_intent),
+                nightmares=c.nightmares,
+            )
+            for c in await repo.list_checkins(s, token)
+        ]
+        results = [r for r in await repo.list_scale_results(s, token) if r.complete]
+        doses = (await s.scalars(select(Dose).where(Dose.token == token, Dose.status != "pending"))).all()
+        cssrs = [r for r in results if r.scale == "cssrs_screen"]
+        last_in = await s.scalar(select(func.max(Message.ts)).where(Message.token == token, Message.direction == "in"))
+        return PatientSnapshot(
+            token=token,
+            as_of=now,
+            age=now.year - profile.birth_year,
+            sex=profile.sex,  # type: ignore[arg-type]
+            modules=set(profile.modules),  # type: ignore[arg-type]
+            therapy_day=(now.date() - profile.therapy_start).days,
+            checkins=checkins,
+            scale_results=[
+                ScalePoint(scale=r.scale, ts=r.ts, total=r.total, item9=r.item9, scheduled=r.scheduled)
+                for r in results
+                if r.scale != "cssrs_screen"
+            ],
+            doses=[DoseRecord(planned_at=d.planned_at, status=d.status) for d in doses],  # type: ignore[arg-type]
+            cssrs_risk=cssrs[-1].severity if cssrs else None,  # type: ignore[arg-type]
+            weapon_access=False,
+            last_patient_message_at=last_in,
+            open_alert_codes=await repo.open_alert_codes(s, token),
+        )
+
+    async def after_data_change(self, token: str) -> list[Action]:
+        async with session.get_clinical_session() as s:
+            actions = evaluate(await self._snapshot(s, token))
+            for a in actions:
+                if a.creates_alert:
+                    assert a.level is not None
+                    await repo.add_alert(
+                        s, token=token, level=a.level, code=a.code, reason=a.reason_ru, ts=clock.now(), params=a.params
+                    )
+        self.fired += [a.code for a in actions]
+        return [a for a in actions if a.type != "alert"]
+
+    async def is_in_crisis(self, token: str) -> bool:
+        async with session.get_clinical_session() as s:
+            alerts = await repo.list_alerts(s, token=token, open_only=True)
+        return any(a.level == "acute" for a in alerts)
+
+
+# ----------------------------------------------------------------------------------------------- fixtures
+
+
+@pytest.fixture
+async def andrey(monkeypatch: pytest.MonkeyPatch) -> AsyncIterator[RuleMonitoring]:
+    monkeypatch.setattr(clock, "get_settings", lambda: Settings(_env_file=None, demo_clock=True))
+    monkeypatch.setattr(integrations, "drug_info", FakeDrugInfo())
+    monkeypatch.setattr(med_qa_module, "get_gateway", lambda: LLMGateway(MockTasks()))
+    clock.set_now(DAY0)
+    await session.configure(identity_url=MEM, clinical_url=MEM)
+    await init_db()
+    now = clock.now()
+    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-smirnova",
+            now=now,
+            birth_year=1992,
+            sex="m",
+        )
+        await identity_repo.record_consent(s, token=TOKEN, version="v1", now=now)
+    async with session.get_clinical_session() as s:
+        await repo.upsert_profile(
+            s,
+            token=TOKEN,
+            doctor_id="doc-smirnova",
+            modules=["veteran", "depression"],
+            therapy_start=now.date(),
+            next_visit=now.date() + timedelta(days=28),
+            birth_year=1992,
+            sex="m",
+        )
+        s.add(
+            Prescription(
+                token=TOKEN, drug="сертралин", dose="50 мг", schedule={"times": ["08:00"]}, start_date=now.date()
+            )
+        )
+        screening = now - timedelta(days=3)  # day −3 screening (persona)
+        for scale, total in (("pcptsd5", 4), ("pcl5", 46), ("phq9", 14), ("auditc", 3)):
+            await repo.add_scale_result(
+                s,
+                token=TOKEN,
+                scale=scale,
+                total=total,
+                severity="-",
+                complete=True,
+                ts=screening,
+                item9=0 if scale == "phq9" else None,
+            )
+        await repo.set_dialog_state(s, token=TOKEN, state="idle", data={"consent": True, "adult": True}, now=now)
+    mon = RuleMonitoring()
+    monitoring_contracts.get_monitoring()  # import lane B's module first if it exists, then override it
+    previous = monitoring_contracts._factory
+    monitoring_contracts.register_monitoring(lambda: mon)
+    yield mon
+    monitoring_contracts.register_monitoring(previous)
+    clock.reset()
+    await session.configure()
+
+
+# ----------------------------------------------------------------------------------------------- helpers
+
+
+class Andrey:
+    def __init__(self) -> None:
+        self.engine = Engine(gateway=LLMGateway(MockTasks()))  # monitoring resolved via register_monitoring
+
+    async def text(self, t: str) -> list[Reply]:
+        return await self.engine.handle(TOKEN, Event(type="text", text=t))
+
+    async def press(self, data: str) -> list[Reply]:
+        return await self.engine.handle(TOKEN, Event(type="button", button=data))
+
+    async def system(self, kind: str, **params: Any) -> list[Reply]:
+        return await self.engine.handle(TOKEN, Event(type="system", system=kind, params=params))  # type: ignore[arg-type]
+
+    async def answer_scale(self, scale_id: str, values: list[int]) -> list[Reply]:
+        out: list[Reply] = []
+        for item, value in zip(load_scale(scale_id).items, values, strict=True):
+            out += await self.press(f"scale:{scale_id}:{item.id}:{value}")
+        return out
+
+
+def buttons(replies: list[Reply]) -> list[str]:
+    return [b.data for r in replies for row in r.buttons for b in row]
+
+
+def joined(replies: list[Reply]) -> str:
+    return "\n".join(r.text for r in replies)
+
+
+async def 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 open_alerts() -> list[Any]:
+    async with session.get_clinical_session() as s:
+        return list(await repo.list_alerts(s, token=TOKEN, open_only=True))
+
+
+async def ack_all() -> None:
+    async with session.get_clinical_session() as s:
+        for a in await repo.list_alerts(s, token=TOKEN, open_only=True):
+            await repo.ack_alert(s, alert_id=a.id, doctor_id="doc-smirnova", ts=clock.now())
+
+
+async def new_dose(day: date) -> int:
+    """What the scheduler (#13) does before sending the reminder: a planned dose row."""
+    async with session.get_clinical_session() as s:
+        rx = await s.scalar(select(Prescription).where(Prescription.token == TOKEN))
+        assert rx is not None
+        row = Dose(token=TOKEN, prescription_id=rx.id, planned_at=clock.now(), status="pending")
+        s.add(row)
+        await s.flush()
+        return row.id
+
+
+# ----------------------------------------------------------------------------------------------- the 13 days
+
+
+async def test_andrey_13_days(andrey: RuleMonitoring) -> None:
+    a = Andrey()
+    meds = load_meds_texts()
+    daily = load_daily_texts()
+    log: dict[int, list[Reply]] = {}
+    colors: list[str] = []
+
+    for day in range(1, 14):
+        taken, mood, sleep, nightmares, alcohol, free_text = TABLE[day]
+        clock.set_now(DAY0 + timedelta(days=day))
+        if day == 6:
+            await ack_all()  # the doctor called Andrey after the day-5 alerts and acknowledged them
+        r: list[Reply] = []
+
+        # --- 08:00 dose reminder
+        dose_id = await new_dose(clock.today())
+        reminder = await a.system("med_reminder", dose_id=dose_id, drug="сертралин", dose="50 мг")
+        assert "сертралин 50 мг" in reminder[0].text
+        assert buttons(reminder) == [f"med:taken:{dose_id}", f"med:skip:{dose_id}", f"med:side_effects:{dose_id}"]
+        if taken:
+            r += await a.press(f"med:taken:{dose_id}")
+        else:
+            skip = await a.press(f"med:skip:{dose_id}")
+            assert f"med:skip_reason:quit:{dose_id}" in buttons(skip)
+            r += skip + await a.press(f"med:skip_reason:quit:{dose_id}")
+
+        # --- day 7: scheduled scales (PCL-5 = 44, PHQ-9 = 13, item 9 = 0) with friendly intros
+        if day == 7:
+            first = await a.system("scale_due", scale="pcl5", scheduled=True)
+            assert daily.scale_intros.scheduled["pcl5"] in first[0].text
+            await a.system("scale_due", scale="phq9", scheduled=True)  # queued behind PCL-5
+            after_pcl = await a.answer_scale("pcl5", [2] * 16 + [3] * 4)
+            assert daily.scale_intros.scheduled["phq9"] in after_pcl[-1].text
+            await a.answer_scale("phq9", [2] * 5 + [1] * 3 + [0])
+            assert (await state())[0] == "idle"
+
+        # --- 20:00 daily check-in
+        clock.set_now(DAY0 + timedelta(days=day, hours=12))
+        start = await a.system("daily_checkin")
+        assert start[0].text == daily.checkin.start and buttons(start)[0] == "daily:mood:0"
+        r += await a.press(f"daily:mood:{mood}")  # dose already answered by the reminder → no med question
+        assert "daily:sleep:<4" in buttons(r[-1:])
+        r += await a.press(f"daily:sleep:{sleep}")
+        r += await a.press(f"daily:nightmares:{'yes' if nightmares else 'no'}")
+        r += await a.press(f"daily:alcohol:{alcohol}")
+        assert buttons(r[-1:]) == ["daily:text:skip"]
+        r += await a.text(free_text) if free_text else await a.press("daily:text:skip")
+        log[day] = r
+        if day == 10:  # «записаться раньше» offered by rule alcohol_3d
+            visit = await a.press("menu:visit")
+            assert visit[0].text == daily.visit.requested and (await state())[1]["visit_requests"]
+
+        if day == 13:
+            break  # the crisis path is checked below
+        assert (await state())[0] == "idle"
+        colors.append(patient_status(AlertLevel[x.level.upper()] for x in await open_alerts()).color)
+
+    # Day 2: first nausea → instruction excerpt (med_qa, section side_effects), not generated advice.
+    assert "По инструкции к препарату" in joined(log[2]) and "проходит в течение 1–2 недель" in joined(log[2])
+    # Day 3: same effect again → short support, the instruction is not repeated.
+    assert meds.side_effects.repeat in joined(log[3]) and "По инструкции" not in joined(log[3])
+    # Day 5: «решил бросить» → rule quit_intent → delayed onset + do not stop on your own + doctor notified.
+    day5 = joined(log[5])
+    assert "2–4 недели" in day5 and "не прекращайте" in day5 and meds.med_info["delayed_effect"].notice in day5
+    assert "200 мг" not in day5  # dosing sentences of the instruction are never quoted
+    # Day 10: sleep < 4 h ×3 + alcohol ×3 → interaction info + «записаться раньше».
+    day10 = joined(log[10])
+    assert "Совместный приём с алкоголем не рекомендуется" in day10
+    assert buttons(log[10])[-1] == "menu:visit"
+    # Day 11: alerts still open → the rule stays silent (no repeated interaction text).
+    assert "Совместный приём с алкоголем" not in joined(log[11]) and "menu:visit" not in buttons(log[11])
+    assert colors == COLORS[:12]
+
+    # Day 13: «всё бессмысленно» → POSSIBLE → C-SSRS screen (buttons only) took over the free text.
+    day13 = log[13]
+    assert day13[-1].kind == "question" and "scale:cssrs_screen:cssrs_1:0" in buttons(day13[-1:])
+    r = await a.answer_scale("cssrs_screen", [0] * 6)
+    # Back in the check-in: the message counts as the answer → check-in closed → mood ≤ 3 three days →
+    # unscheduled PHQ-9 with its own intro.
+    assert daily.checkin.done in joined(r) and daily.scale_intros.unscheduled["phq9"] in r[-1].text
+    st, data = await state()
+    assert st == "scale.phq9" and data["scales"]["phq9"]["scheduled"] is False
+    r = await a.answer_scale("phq9", [2] * 8 + [2])  # total 18, item 9 = 2
+    assert any(x.kind == "crisis" for x in r) and any(x.kind == "safety_plan" for x in r)
+    st, data = await state()
+    # C-SSRS was completed minutes ago: item 9 opens the crisis but does not ask the same screen again (#56).
+    assert data["crisis"]["open"] is True and st == "crisis" and not data.get("pending_scales")
+    async with session.get_clinical_session() as s:
+        assert len(await repo.list_scale_results(s, TOKEN, "cssrs_screen")) == 1
+    acute = [x for x in await open_alerts() if x.level == "acute"]
+    assert [x.code for x in acute] == ["phq9_item9"]
+    assert patient_status(AlertLevel[x.level.upper()] for x in await open_alerts()).color == COLORS[12]
+
+    # Stored check-ins match the persona table.
+    async with session.get_clinical_session() as s:
+        rows = {c.day: c for c in await repo.list_checkins(s, TOKEN)}
+        doses = (await s.scalars(select(Dose).where(Dose.token == TOKEN).order_by(Dose.id))).all()
+        phq = list(await repo.list_scale_results(s, TOKEN, "phq9"))
+    for day, (taken, mood, sleep, nightmares, alcohol, _text) in TABLE.items():
+        c = rows[(DAY0 + timedelta(days=day)).date()]
+        assert (c.med_taken, c.mood, c.sleep_bucket, c.nightmares, c.alcohol_bucket) == (
+            taken,
+            mood,
+            sleep,
+            nightmares,
+            alcohol,
+        ), day
+    for day in (2, 3, 4):
+        assert "nausea" in rows[(DAY0 + timedelta(days=day)).date()].side_effects
+    assert rows[(DAY0 + timedelta(days=5)).date()].quit_intent is True
+    assert rows[(DAY0 + timedelta(days=13)).date()].free_text_scrubbed is None  # taken over by the C-SSRS
+    assert [d.status for d in doses] == ["skipped" if d == 5 else "taken" for d in range(1, 14)]
+    assert [(p.total, p.item9, p.scheduled) for p in phq] == [(14, 0, True), (13, 0, True), (18, 2, False)]
+    for code in ("side_effect_3d", "quit_intent", "sleep_lt4_3n", "alcohol_3d", "mood_low_3d", "phq9_item9"):
+        assert code in andrey.fired, code

tests/test_dialog_veteran_onboarding.py

@@ -0,0 +1,368 @@
+"""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 in [x.text for x in r]
+
+
+# ----------------------------------------------------------------------------------------------- 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}»"

tests/test_scheduler_demo.py

@@ -0,0 +1,167 @@
+"""Demo scheduler (#13) on the virtual clock: real dialog engine, real monitoring, sqlite."""
+
+from collections.abc import AsyncIterator
+from datetime import datetime, timedelta
+
+import pytest
+from sqlalchemy import select
+
+from app.config import Settings
+from app.core import clock
+from app.core.types import ScrubbedText
+from app.db import repo, session
+from app.db.bootstrap import init_db
+from app.db.clinical import Dose, Prescription
+from app.dialog import engine as _engine  # noqa: F401  (registers flows in order)
+from app.dialog.flows.veteran_daily import load_daily_texts
+from app.integrations.base import MemoryOutbox, integrations
+from app.scheduler import demo, jobs
+
+TOKEN = "b" * 32
+MEM = "sqlite+aiosqlite:///:memory:"
+DAY0 = datetime(2026, 10, 5, 8, 0)  # visit day (Europe/Moscow)
+
+
+@pytest.fixture
+async def outbox(monkeypatch: pytest.MonkeyPatch) -> AsyncIterator[MemoryOutbox]:
+    monkeypatch.setattr(clock, "get_settings", lambda: Settings(_env_file=None, demo_clock=True))
+    box = MemoryOutbox()
+    monkeypatch.setattr(integrations, "outbox", box)
+    clock.set_now(DAY0)
+    await session.configure(identity_url=MEM, clinical_url=MEM)
+    await init_db()
+    now = clock.now()
+    async with session.get_clinical_session() as s:
+        await repo.upsert_profile(
+            s,
+            token=TOKEN,
+            doctor_id="doc-smirnova",
+            modules=["veteran"],
+            therapy_start=now.date(),
+            birth_year=1992,
+            sex="m",
+        )
+        s.add(Prescription(token=TOKEN, drug="сертралин", dose="50 мг", schedule={"times": ["08:00"]}))
+        for scale, total in (("pcl5", 46), ("phq9", 14), ("auditc", 3)):
+            await repo.add_scale_result(
+                s,
+                token=TOKEN,
+                scale=scale,
+                total=total,
+                severity="-",
+                complete=True,
+                ts=now - timedelta(days=3),
+                item9=0 if scale == "phq9" else None,
+            )
+        await repo.add_message(s, token=TOKEN, direction="in", text=ScrubbedText("..."), ts=now)
+        await repo.set_dialog_state(s, token=TOKEN, state="idle", data={"consent": True, "adult": True}, now=now)
+    yield box
+    clock.reset()
+    await session.configure()
+
+
+async def at(day: int, hh: int, mm: int = 0) -> list[demo.Job]:
+    clock.set_now(DAY0.replace(hour=hh, minute=mm) + timedelta(days=day))
+    return await demo.run_until(clock.now())
+
+
+def kinds(delivered: list[demo.Job], *, evaluation: bool = False) -> list[str]:
+    """Delivered job kinds; the daily monitoring pass (from 09:00) is left out unless asked for."""
+    return [j.kind for j in delivered if evaluation or j.kind != "evaluation"]
+
+
+async def doses() -> list[Dose]:
+    async with session.get_clinical_session() as s:
+        return list((await s.scalars(select(Dose).order_by(Dose.planned_at))).all())
+
+
+async def dialog() -> tuple[str, dict]:
+    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 test_reminder_once_then_missed(outbox: MemoryOutbox) -> None:
+    assert kinds(await at(1, 8, 5)) == ["med_reminder"]
+    [dose] = await doses()
+    assert dose.status == "pending"
+    sent = outbox.drain(TOKEN)
+    assert any(b.data == f"med:taken:{dose.id}" for r in sent for row in r.buttons for b in row)
+    assert kinds(await at(1, 9)) == []  # idempotent: the same reminder is not sent again
+    assert kinds(await at(1, 10, 6)) == ["missed"]
+    assert [d.status for d in await doses()] == ["missed"]
+    assert kinds(await at(1, 11)) == []
+
+
+async def test_answered_dose_is_not_missed(outbox: MemoryOutbox) -> None:
+    from app.dialog.contracts import Event, get_engine
+
+    await at(1, 8)
+    [dose] = await doses()
+    await get_engine().handle(TOKEN, Event(type="button", button=f"med:taken:{dose.id}"))
+    assert kinds(await at(1, 10, 30)) == []
+    assert [d.status for d in await doses()] == ["taken"]
+
+
+async def test_three_missed_days_alert_and_reach_out(outbox: MemoryOutbox) -> None:
+    for day in (1, 2, 3):
+        await at(day, 8)
+        await at(day, 10, 1)
+    outbox.drain(TOKEN)
+    async with session.get_clinical_session() as s:
+        codes = [a.code for a in await repo.list_alerts(s, token=TOKEN)]
+    assert "missed_doses_7d" in codes
+
+
+async def test_stale_events_are_not_sent(outbox: MemoryOutbox) -> None:
+    # The clock jumped straight to the evening: no reminder for 08:00, no dose row, only the check-in.
+    delivered = await at(1, 20, 5)
+    assert kinds(delivered) == ["daily_checkin"]
+    assert await doses() == []
+    assert outbox.drain(TOKEN)[0].text == load_daily_texts().checkin.start
+    assert (await dialog())[0].startswith("daily.")
+    assert kinds(await at(1, 21)) == []
+
+
+async def test_scales_by_module_schedule(outbox: MemoryOutbox) -> None:
+    assert "scale_due" not in kinds(await at(7, 10, 5))  # screening on day −3: PCL-5/PHQ-9 every 14 days
+    delivered = await at(11, 10, 5)
+    assert [j.params["scale"] for j in delivered if j.kind == "scale_due"] == ["pcl5", "phq9"]
+    state, data = await dialog()
+    assert state == "scale.pcl5" and "phq9" in str(data.get("pending_scales"))
+
+
+async def test_silence_evaluation(outbox: MemoryOutbox) -> None:
+    # Last message: day 0, 08:00. Day 2 — not yet; day 3, 09:05 — silent for more than 3 days.
+    assert kinds(await at(2, 9, 5), evaluation=True) == ["med_reminder", "evaluation"]
+    async with session.get_clinical_session() as s:
+        assert not await repo.list_alerts(s, token=TOKEN)
+    outbox.drain(TOKEN)
+    assert "evaluation" in kinds(await at(3, 9, 5), evaluation=True)
+    async with session.get_clinical_session() as s:
+        assert [a.code for a in await repo.list_alerts(s, token=TOKEN)] == ["no_contact_3d"]
+    # caring contact pushed to the patient (system:reach_out)
+    assert any("как вы" in r.text.lower() for r in outbox.drain(TOKEN))
+
+
+async def test_no_contact_without_consent_or_in_crisis(outbox: MemoryOutbox) -> None:
+    async with session.get_clinical_session() as s:
+        await repo.set_dialog_state(
+            s,
+            token=TOKEN,
+            state="crisis",
+            data={"consent": True, "adult": True, "crisis": {"open": True}},
+            now=clock.now(),
+        )
+    assert await at(1, 8) == []  # crisis: no reminder
+    async with session.get_clinical_session() as s:
+        await repo.set_dialog_state(s, token=TOKEN, state="idle", data={"consent": False}, now=clock.now())
+    assert await at(1, 8, 30) == []
+
+
+async def test_jobs_entrypoint_runs_demo_scheduler(outbox: MemoryOutbox) -> None:
+    clock.set_now(DAY0 + timedelta(days=1))
+    await jobs.run_until_now()
+    assert [d.status for d in await doses()] == ["pending"]
+    assert outbox.drain(TOKEN)

tests/test_services_monitoring.py

@@ -0,0 +1,279 @@
+"""MonitoringService (#38): snapshot from the clinical DB → real rules → alerts, colour cache, audit, notify."""
+
+from collections.abc import AsyncIterator
+from datetime import datetime, timedelta
+from typing import Any
+
+import pytest
+from sqlalchemy import select
+
+from app.config import Settings
+from app.core import clock
+from app.core.types import ScrubbedText
+from app.db import repo, session
+from app.db.audit import AuditLog
+from app.db.bootstrap import init_db
+from app.db.clinical import Dose, Prescription
+from app.integrations.base import MemoryNotifier, integrations
+from app.services import contracts, monitoring
+from app.services.snapshot import build_snapshot
+
+TOKEN = "a" * 32
+DOCTOR = "doc-smirnova"
+MEM = "sqlite+aiosqlite:///:memory:"
+DAY0 = datetime(2026, 10, 5, 8, 0)
+
+# Persona (docs/product/01_demo_persona.md): day → (dose taken, mood, sleep, nightmares, alcohol, side effects,
+# quit intent). Same table as tests/test_dialog_veteran_days.py, written straight to the DB.
+TABLE: dict[int, tuple[bool, int, str, bool, str, list[str], bool]] = {
+    1: (True, 4, "4-6", True, "none", [], False),
+    2: (True, 4, "4-6", True, "none", ["nausea"], False),
+    3: (True, 4, "4-6", True, "none", ["nausea"], False),
+    4: (True, 3, "4-6", False, "none", ["nausea"], False),
+    5: (False, 3, "4-6", False, "none", [], True),
+    6: (True, 4, "6-8", False, "none", [], False),
+    7: (True, 4, "6-8", False, "none", [], False),
+    8: (True, 4, "<4", True, "1-2", [], False),
+    9: (True, 4, "<4", True, "more", [], False),
+    10: (True, 4, "<4", True, "more", [], False),
+    11: (True, 3, "4-6", False, "1-2", [], False),
+    12: (True, 3, "4-6", False, "none", [], False),
+    13: (True, 2, "<4", True, "none", [], False),
+}
+COLORS = ["green"] * 3 + ["yellow"] * 2 + ["green"] * 4 + ["yellow"] * 3 + ["red"]
+
+
+class FailingNotifier:
+    async def send(self, *args: Any, **kwargs: Any) -> None:
+        raise ConnectionError("channel down")
+
+
+@pytest.fixture
+async def db(monkeypatch: pytest.MonkeyPatch) -> AsyncIterator[MemoryNotifier]:
+    monkeypatch.setattr(clock, "get_settings", lambda: Settings(_env_file=None, demo_clock=True))
+    notifier = MemoryNotifier()
+    monkeypatch.setattr(integrations, "notifier", notifier)
+    clock.set_now(DAY0)
+    await session.configure(identity_url=MEM, clinical_url=MEM)
+    await init_db()
+    async with session.get_clinical_session() as s:
+        await repo.upsert_profile(
+            s,
+            token=TOKEN,
+            doctor_id=DOCTOR,
+            modules=["veteran", "depression"],
+            therapy_start=clock.today(),
+            birth_year=1992,
+            sex="m",
+        )
+        s.add(Prescription(token=TOKEN, drug="сертралин", dose="50 мг", schedule={"times": ["08:00"]}))
+        for scale, total in (("pcl5", 46), ("phq9", 14), ("auditc", 3)):
+            await repo.add_scale_result(
+                s,
+                token=TOKEN,
+                scale=scale,
+                total=total,
+                severity="-",
+                complete=True,
+                ts=clock.now() - timedelta(days=3),
+                item9=0 if scale == "phq9" else None,
+            )
+        await repo.set_dialog_state(s, token=TOKEN, state="idle", data={}, now=clock.now())
+    yield notifier
+    clock.register_persistence(None, None)
+    clock.reset()
+    await session.configure()
+
+
+async def scale(
+    name: str, total: int, *, item9: int | None = None, severity: str = "-", scheduled: bool = True
+) -> None:
+    async with session.get_clinical_session() as s:
+        await repo.add_scale_result(
+            s,
+            token=TOKEN,
+            scale=name,
+            total=total,
+            severity=severity,
+            complete=True,
+            ts=clock.now(),
+            item9=item9,
+            scheduled=scheduled,
+        )
+
+
+async def color() -> str | None:
+    async with session.get_clinical_session() as s:
+        profile = await repo.get_profile(s, TOKEN)
+        assert profile is not None
+        return profile.status_color
+
+
+async def ack_all() -> None:
+    async with session.get_clinical_session() as s:
+        for a in await repo.list_alerts(s, token=TOKEN, open_only=True):
+            await repo.ack_alert(s, alert_id=a.id, doctor_id=DOCTOR, ts=clock.now())
+        await monitoring.refresh_status(s, TOKEN)
+
+
+async def live_day(day: int) -> None:
+    """What the dialog writes on one persona day: the dose, an incoming message and the evening check-in."""
+    taken, mood, sleep, nightmares, alcohol, effects, quit_intent = TABLE[day]
+    clock.set_now(DAY0 + timedelta(days=day))
+    async with session.get_clinical_session() as s:
+        rx = await s.scalar(select(Prescription))
+        assert rx is not None
+        s.add(Dose(token=TOKEN, prescription_id=rx.id, planned_at=clock.now(), status="taken" if taken else "skipped"))
+        await repo.add_message(s, token=TOKEN, direction="in", text=ScrubbedText("..."), ts=clock.now())
+        await repo.upsert_checkin(
+            s,
+            token=TOKEN,
+            day=clock.today(),
+            now=clock.now(),
+            mood=mood,
+            med_taken=taken,
+            sleep_bucket=sleep,
+            alcohol_bucket=alcohol,
+            nightmares=nightmares,
+            side_effects=effects,
+            quit_intent=quit_intent,
+        )
+
+
+async def test_service_is_registered() -> None:
+    assert isinstance(contracts.get_monitoring(), monitoring.RuleMonitoring)
+
+
+async def test_andrey_13_days(db: MemoryNotifier) -> None:
+    mon = monitoring.RuleMonitoring()
+    colors: list[str | None] = []
+    returned: dict[int, list[str]] = {}
+    for day in range(1, 14):
+        if day == 6:
+            await ack_all()  # the doctor called Andrey after the day-5 alerts
+        await live_day(day)
+        if day == 7:
+            await scale("pcl5", 44)
+            await scale("phq9", 13, item9=0)
+        actions = await mon.after_data_change(TOKEN)
+        if day == 13:
+            assert [a.code for a in actions] == ["mood_low_3d"]  # dialog action: unscheduled PHQ-9
+            await scale("cssrs_screen", 0, severity="none", scheduled=False)
+            await scale("phq9", 18, item9=2, scheduled=False)
+            actions += await mon.after_data_change(TOKEN)
+        returned[day] = [a.type for a in actions]
+        colors.append(await color())
+
+    assert colors == COLORS
+    assert "med_info" in returned[5] and "alert" not in returned[5]  # alerts are persisted, not returned
+    assert {"med_info", "suggest_visit"} <= set(returned[10])
+    assert {"crisis", "show_safety_plan", "schedule_scale"} <= set(returned[13])
+    async with session.get_clinical_session() as s:
+        alerts = list(await repo.list_alerts(s, token=TOKEN))
+        crisis_audit = (await s.scalars(select(AuditLog).where(AuditLog.action == "crisis_acute"))).all()
+    codes = [a.code for a in reversed(alerts)]
+    assert codes == [
+        "side_effect_3d",
+        "quit_intent",
+        "sleep_lt4_3n",
+        "alcohol_3d",
+        "phq9_item9",  # same evaluation: YAML order
+        "phq9_rise",
+    ]
+    assert len(crisis_audit) == 1 and crisis_audit[0].token == TOKEN and crisis_audit[0].actor == "system"
+    acute = [(d, lvl) for d, lvl, _text, _tok in db.sent if lvl == "acute"]
+    assert acute == [(DOCTOR, "acute"), (monitoring.ON_DUTY_DOCTOR, "acute")]
+    assert all(tok == TOKEN for *_rest, tok in db.sent)
+    assert await mon.is_in_crisis(TOKEN)
+
+
+async def test_running_crisis_is_not_restarted(db: MemoryNotifier) -> None:
+    mon = monitoring.RuleMonitoring()
+    await scale("phq9", 18, item9=1)
+    first = await mon.after_data_change(TOKEN)
+    assert any(a.type == "crisis" for a in first)
+    clock.advance(hours=1)
+    again = await mon.after_data_change(TOKEN)
+    assert not any(a.code == "phq9_item9" for a in again)
+    async with session.get_clinical_session() as s:
+        assert [a.code for a in await repo.list_alerts(s, token=TOKEN, open_only=True) if a.level == "acute"] == [
+            "phq9_item9"
+        ]
+    await ack_all()
+    assert not await mon.is_in_crisis(TOKEN)
+    assert await color() == "green"
+
+
+async def test_old_cssrs_high_does_not_restart_crisis(db: MemoryNotifier) -> None:
+    mon = monitoring.RuleMonitoring()
+    await scale("cssrs_screen", 4, severity="high", scheduled=False)
+    assert any(a.type == "crisis" for a in await mon.after_data_change(TOKEN))
+    await ack_all()
+    clock.advance(days=2)
+    assert not any(a.type == "crisis" for a in await mon.after_data_change(TOKEN))
+
+
+async def test_weapon_access_from_dialog_facts(db: MemoryNotifier) -> None:
+    async with session.get_clinical_session() as s:
+        await repo.set_dialog_state(
+            s, token=TOKEN, state="idle", data={"facts": {"weapon_access": True}}, now=clock.now()
+        )
+        snap = await build_snapshot(s, TOKEN, clock.now())
+    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"
+
+
+async def test_snapshot_mapping_and_unknown_patient(db: MemoryNotifier) -> None:
+    clock.set_now(DAY0 + timedelta(days=4))
+    await live_day(4)
+    async with session.get_clinical_session() as s:
+        snap = await build_snapshot(s, TOKEN, clock.now())
+        assert await build_snapshot(s, "f" * 32, clock.now()) is None
+    assert snap is not None
+    assert (snap.age, snap.sex, snap.therapy_day, snap.modules) == (34, "m", 4, {"veteran", "depression"})
+    c = snap.checkins[-1]
+    assert (c.sleep_hours_bucket, c.alcohol, c.side_effects) == ("4-6", "none", ["nausea"])
+    assert [p.scale for p in snap.scale_results] == ["pcl5", "phq9", "auditc"]
+    assert snap.last_patient_message_at is not None and snap.last_patient_message_at.date() == clock.today()
+    assert await monitoring.RuleMonitoring().after_data_change("f" * 32) == []
+
+
+async def test_young_adult_enhanced_monitoring(db: MemoryNotifier) -> None:
+    async with session.get_clinical_session() as s:
+        await repo.upsert_profile(s, token=TOKEN, birth_year=2004)
+    actions = await monitoring.RuleMonitoring().after_data_change(TOKEN)
+    assert not any(a.type == "enhanced_monitoring" for a in actions)  # applied by the service itself
+    async with session.get_clinical_session() as s:
+        profile = await repo.get_profile(s, TOKEN)
+        assert profile is not None and profile.enhanced_monitoring is True
+        assert not await repo.list_alerts(s, token=TOKEN)
+    assert await color() == "green"
+
+
+async def test_notifier_failure_keeps_alert(db: MemoryNotifier, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(integrations, "notifier", FailingNotifier())
+    await scale("pcl5", 55)
+    await monitoring.RuleMonitoring().after_data_change(TOKEN)
+    async with session.get_clinical_session() as s:
+        assert [(a.code, a.level) for a in await repo.list_alerts(s, token=TOKEN)] == [("pcl5_severe", "high")]
+    assert await color() == "yellow"
+
+
+async def test_clock_offset_persists(db: MemoryNotifier) -> None:
+    clock.reset()
+    await monitoring.init_clock_persistence()
+    clock.advance(days=3)
+    await monitoring.flush_clock_writes()
+    async with session.get_clinical_session() as s:
+        assert await repo.load_clock_offset(s) == pytest.approx(timedelta(days=3).total_seconds())
+
+    # "Restart": in-memory offset lost, restored from clock_state.
+    clock.register_persistence(None, None)
+    clock.reset()
+    await monitoring.init_clock_persistence()
+    assert clock.offset() == timedelta(days=3)