Merge pull request 'feat(questionnaire): revised intake, diaries and follow-up monitoring' (#53) from feat/52-questionnaire into dev

Reviewed-on: #53

app/api/questionnaire.py

@@ -0,0 +1,98 @@
+"""Audited questionnaire review for the synthetic prototype.
+
+Like the existing API, doctor_id is supplied by the caller; production must put
+these routes behind authenticated staff identity. No clinical data cross into identity.
+"""
+
+from typing import Literal
+
+from fastapi import APIRouter, HTTPException
+from pydantic import BaseModel, ConfigDict, Field
+
+from app.core import clock
+from app.db import repo
+from app.db.audit import audit
+from app.db.intake import QuestionnaireRecord
+from app.db.session import get_clinical_session
+from app.services import questionnaire as service
+
+router = APIRouter(tags=["questionnaire-review"])
+
+
+class ReviewIn(BaseModel):
+    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)
+    doctor_id: str = Field(min_length=1)
+    assessment_performed: Literal[True]
+    decision: Literal["resume", "keep_hold"]
+    reference: str = Field(min_length=1, max_length=200, pattern=r"^[A-Za-z0-9_./:#-]+$")
+
+
+async def require_doctor(session, token, doctor_id):
+    profile = await repo.get_profile(session, token)
+    if profile is None or profile.doctor_id != doctor_id:
+        raise HTTPException(404, "patient not found")
+
+
+@router.get("/patients/{token}/questionnaires")
+async def history(token: str, doctor_id: str):
+    async with get_clinical_session() as session:
+        await require_doctor(session, token, doctor_id)
+        rows = await service.records(session, token)
+        await audit(session, actor=doctor_id, action="view_questionnaires", token=token, ts=clock.now())
+        return [
+            {
+                "id": r.id,
+                "kind": r.kind,
+                "status": r.status,
+                "version": r.version,
+                "policy_version": r.policy_version,
+                "started_at": r.started_at,
+                "completed_at": r.completed_at,
+                "answers": r.answers,
+                "details": r.details,
+            }
+            for r in rows
+        ]
+
+
+@router.post("/patients/{token}/questionnaires/{record_id}/review")
+async def review(token: str, record_id: str, body: ReviewIn):
+    now = clock.now()
+    async with get_clinical_session() as session:
+        await service.lock_patient(session, token)
+        await require_doctor(session, token, body.doctor_id)
+        state = await repo.get_dialog_state(session, token)
+        row = await session.get(QuestionnaireRecord, record_id)
+        if row is None or row.token != token:
+            raise HTTPException(404, "record not found")
+        if not state or state.data.get("consent") is not True:
+            raise HTTPException(409, "consent required")
+        if row.status not in ("review_required", "interrupted"):
+            raise HTTPException(409, "record does not require review")
+        if body.decision == "resume" and (state.data.get("crisis") or {}).get("open"):
+            raise HTTPException(409, "close the acute episode after assessment before resuming routine work")
+        detail = {
+            "doctor_id": body.doctor_id,
+            "decision": body.decision,
+            "reference": body.reference,
+            "at": now.isoformat(),
+        }
+        row.details = {**row.details, "clinician_review": detail}
+        if body.decision == "resume":
+            row.status = "reviewed"
+            remaining = [
+                r
+                for r in await service.records(session, token)
+                if r.id != row.id and r.status in ("review_required", "interrupted")
+            ]
+            if not remaining:
+                state.data = {**state.data, "intake_review_required": False, "questionnaire_record": None}
+        await audit(
+            session,
+            actor=body.doctor_id,
+            action="review_questionnaire",
+            token=token,
+            details={"record_id": row.id, "decision": body.decision, "reference": body.reference},
+            ts=now,
+        )
+        return {"id": row.id, "status": row.status, "review": detail}

app/api/routes.py

@@ -7,6 +7,7 @@ Lane B replaces bodies in #12 without changing paths or schemas (those are the f
 from fastapi import APIRouter, HTTPException
 
 from app.api import fixtures
+from app.api.questionnaire import router as questionnaire_router
 from app.api.schemas import (
     AckIn,
     AlertOut,
@@ -26,6 +27,7 @@ from app.dialog.contracts import get_engine
 from app.integrations.base import MemoryOutbox, integrations
 
 router = APIRouter(prefix="/api/v1")
+router.include_router(questionnaire_router)
 
 
 # ------------------------------------------------------------------ doctor dashboard

app/core/cssrs_content.py

@@ -0,0 +1,59 @@
+"""Russian Recent source transcription, separate from the legacy sequential scale.
+
+This content does not authorize administering or continuing the questionnaire.
+Use questionnaire.score_cssrs_recent for required/missing items and risk, and
+evaluate_intake for the clinical policy gate before presenting another question.
+Never render interviewer guidance as if it were another patient question.
+"""
+
+from pathlib import Path
+from typing import Literal
+
+import yaml
+from pydantic import BaseModel, ConfigDict, Field, model_validator
+
+from app.config import get_settings
+
+ItemId = Literal["q1", "q2", "q3", "q4", "q5", "q6_lifetime", "q6_recent"]
+
+
+class SourceModel(BaseModel):
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+
+class CssrsOption(SourceModel):
+    value: int = Field(strict=True, ge=0, le=1)
+    label: str
+
+
+class CssrsItem(SourceModel):
+    id: ItemId
+    text: str = Field(min_length=1)
+    interviewer_guidance: str | None
+    recall_period: Literal["past_month", "lifetime", "past_3_months"]
+    yes_risk: Literal["low", "moderate", "high"]
+
+
+class CssrsRussianForm(SourceModel):
+    id: Literal["cssrs_recent_ru"]
+    version: str = Field(min_length=1)
+    title: str
+    subtitle: Literal["Screen Version - Recent"]
+    period_label_original: str
+    initial_instruction: str
+    branch_instruction: str
+    options: tuple[CssrsOption, CssrsOption]
+    items: tuple[CssrsItem, ...]
+
+    @model_validator(mode="after")
+    def _structure(self) -> "CssrsRussianForm":
+        if tuple(item.id for item in self.items) != ("q1", "q2", "q3", "q4", "q5", "q6_lifetime", "q6_recent"):
+            raise ValueError("Recent form requires seven ordered answer fields")
+        if tuple(option.value for option in self.options) != (1, 0):
+            raise ValueError("source answer columns are Yes, No")
+        return self
+
+
+def load_cssrs_russian(path: Path | None = None) -> CssrsRussianForm:
+    path = path or get_settings().config_dir / "instruments" / "cssrs_recent_ru.yaml"
+    return CssrsRussianForm.model_validate(yaml.safe_load(path.read_text(encoding="utf-8")))

app/core/intake_questions.py

@@ -0,0 +1,141 @@
+"""Question selection and exact versioned content, with no side effects."""
+
+from dataclasses import dataclass
+
+import yaml
+
+from app.config import get_settings
+from app.core.cssrs_content import load_cssrs_russian
+from app.core.questionnaire import IntakePolicy, alcohol_expansion, score_alcohol, score_cssrs_recent
+from app.core.scales import ScaleDef, load_scale
+
+
+@dataclass(frozen=True)
+class Question:
+    id: str
+    text: str
+    options: tuple[tuple[object, str], ...] = ()
+    period: str = "context"
+    numeric: bool = False
+    optional: bool = False
+
+
+def content() -> dict:
+    return yaml.safe_load((get_settings().config_dir / "instruments/intake_forms.yaml").read_text())
+
+
+def texts() -> dict:
+    return yaml.safe_load((get_settings().config_dir / "texts/intake.yaml").read_text())
+
+
+def scale_question(scale_id: str, item_id: str) -> Question:
+    forms = content()["forms"]
+    scale = ScaleDef.model_validate(forms[scale_id]) if scale_id in forms else load_scale(scale_id)
+    item = next(i for i in scale.items if i.id == item_id)
+    instruction = scale.instruction or ""
+    # The beverage table is part of the approved questionnaire source.
+    if item_id == "rus_audit_2":
+        instruction += "\n" + texts()["standard_drinks"]
+    return Question(
+        item_id,
+        instruction + "\n\n" + item.text,
+        tuple((o.value, o.label) for o in item.options),
+        scale.recall_period or "",
+    )
+
+
+def group_answers(answers: dict, prefix: str) -> dict[str, int]:
+    return {k: v for k, v in answers.items() if k.startswith(prefix + "_")}
+
+
+def cssrs_answers(answers: dict) -> dict[str, int]:
+    return {k.removeprefix("cssrs_"): v for k, v in answers.items() if k.startswith("cssrs_")}
+
+
+def next_intake(answers: dict, policy: IntakePolicy, sex: str | None) -> Question | None:
+    c = score_cssrs_recent(cssrs_answers(answers))
+    if c.missing:
+        form = load_cssrs_russian()
+        item = next(i for i in form.items if i.id == c.missing[0])
+        period = texts()["periods"][item.recall_period]
+        # Guidance is interviewer-facing; retained in the source and record version.
+        return Question(
+            "cssrs_" + item.id,
+            period + "\n\n" + item.text,
+            tuple((o.value, o.label) for o in form.options),
+            item.recall_period,
+        )
+    for item in ("phq9_1", "phq9_2"):
+        if item not in answers:
+            return scale_question("phq9", item)
+    if "trauma" not in answers:
+        return Question(
+            "trauma", content()["trauma_introduction"] + "\n\n" + content()["trauma_filter"], ((1, "Да"), (0, "Нет"))
+        )
+    if answers["trauma"]:
+        for item in load_scale("tsq").item_ids:
+            if item not in answers:
+                return scale_question("tsq", item)
+        if "recent_trauma" not in answers:
+            return Question("recent_trauma", texts()["recent_trauma"], ((1, "Да"), (0, "Нет")))
+    for item in load_scale("rus_audit_s").item_ids:
+        if item not in answers:
+            return scale_question("rus_audit_s", item)
+    if sex is None and "sex" not in answers:
+        return Question("sex", texts()["sex"], (("f", "Женский"), ("m", "Мужской"), ("unknown", "Не указывать")))
+    for key in ("tbi", "psychotropics"):
+        if key not in answers:
+            return Question(key, texts()[key], ((1, "Да"), (0, "Нет")))
+    if answers["phq9_1"] + answers["phq9_2"] >= policy.phq2_expand_at:
+        for item in load_scale("phq9").item_ids:
+            if item not in answers:
+                return scale_question("phq9", item)
+    if answers["trauma"] and sum(group_answers(answers, "tsq").values()) >= policy.tsq_expand_at:
+        for item in load_scale("pcl5").item_ids:
+            if item not in answers:
+                return scale_question("pcl5", item)
+    alcohol = group_answers(answers, "rus_audit")
+    selected_sex = sex or (answers.get("sex") if answers.get("sex") in ("m", "f") else None)
+    short = score_alcohol(
+        {k: v for k, v in alcohol.items() if k in load_scale("rus_audit_s").item_ids}, short=True, sex=selected_sex
+    )
+    if alcohol_expansion(short, policy):
+        for item in load_scale("rus_audit").item_ids:
+            if item == "rus_audit_2" and alcohol.get("rus_audit_1") == 0:
+                continue
+            if item not in answers:
+                return scale_question("rus_audit", item)
+    return None
+
+
+def diary_questions(
+    modules: list[str], *, suicide_monitoring: bool, weekday: int, day_index: int | None = None
+) -> tuple[Question, ...]:
+    config = yaml.safe_load((get_settings().config_dir / "instruments/diary.yaml").read_text())
+    selected = ["core"]
+    detailed = [name for name in ("veteran", "depression") if name in modules]
+    if len(detailed) == 2:
+        detailed = [detailed[(day_index if day_index is not None else weekday) % 2]]
+    if "veteran" in detailed:
+        selected.append("ptsd")
+    if "depression" in detailed:
+        selected.append("depression")
+    if suicide_monitoring or "depression" in modules:
+        selected.append("suicide")
+    questions = []
+    for section in selected:
+        for item in config[section]:
+            if item.get("weekly") and weekday != 0:
+                continue
+            questions.append(
+                Question(
+                    item["id"],
+                    item["text"],
+                    tuple((o["value"], o["label"]) for o in item.get("options", [])),
+                    "past_week" if item.get("weekly") else "past_day",
+                    item.get("numeric", False),
+                    item.get("optional", False),
+                )
+            )
+    questions.sort(key=lambda q: q.id != "death_thoughts")
+    return tuple(questions)

app/core/intake_runtime.py

@@ -0,0 +1,153 @@
+"""Versioned release gate and deterministic follow-up planning."""
+
+from calendar import monthrange
+from datetime import date, timedelta
+from pathlib import Path
+from typing import Literal
+from zoneinfo import ZoneInfo
+
+import yaml
+from pydantic import BaseModel, ConfigDict, Field, model_validator
+
+from app.config import get_settings
+from app.core.questionnaire import Approval, load_policy
+
+
+class RuntimePolicy(BaseModel):
+    model_config = ConfigDict(extra="forbid", frozen=True)
+    version: str
+    enabled: bool = False
+    approval: Approval | None = None
+    content_review: Approval | None = None
+    contact_review: Approval | None = None
+    timezone: str
+    daily_hour: int = Field(ge=0, le=23)
+    mood_baseline_days: int = Field(ge=2)
+    mood_bad_days: int = Field(ge=1)
+    mood_drop: int = Field(ge=1, le=10)
+    sleep_bad_days: int = Field(ge=1)
+    sleep_hours_below: float = Field(gt=0, le=24)
+    nightmares_at: int = Field(ge=0, le=4)
+    missed_medication_days: int = Field(ge=1)
+    high_risk_silence_days: int = Field(ge=1)
+    medication_days: tuple[int, ...]
+    medication_repeat_days: int = Field(ge=1)
+    route_intervals: dict[str, dict[str, int]]
+    alcohol_repeat_days: int = Field(ge=1)
+    phq9_retriage_delta: int = Field(ge=1)
+    pcl5_retriage_delta: int = Field(ge=1)
+
+    @model_validator(mode="after")
+    def valid(self):
+        ZoneInfo(self.timezone)
+        if set(self.route_intervals) != {"after_crisis", "specialist", "self_help"} or any(
+            set(row) != {"phq9", "pcl5"} for row in self.route_intervals.values()
+        ):
+            raise ValueError("all route and scale intervals must be defined")
+        if not self.medication_days or tuple(sorted(set(self.medication_days))) != self.medication_days:
+            raise ValueError("medication days must be nonempty and strictly increasing")
+        if self.medication_days[0] < 1 or any(v < 1 for row in self.route_intervals.values() for v in row.values()):
+            raise ValueError("intervals must be positive")
+        if any(
+            a and a.policy_version != self.version for a in (self.approval, self.content_review, self.contact_review)
+        ):
+            raise ValueError("review must match the runtime version")
+        return self
+
+    @property
+    def released(self) -> bool:
+        return self.enabled and all((self.approval, self.content_review, self.contact_review))
+
+
+def load_runtime(path: Path | None = None) -> RuntimePolicy:
+    path = path or get_settings().config_dir / "intake_runtime.yaml"
+    return RuntimePolicy.model_validate(yaml.safe_load(path.read_text()))
+
+
+def intake_enabled() -> bool:
+    return load_runtime().released and load_policy().approval is not None
+
+
+def month_before(day: date, months: int = 1) -> date:
+    year, month = divmod(day.year * 12 + day.month - 1 - months, 12)
+    return day.replace(year=year, month=month + 1, day=min(day.day, monthrange(year, month + 1)[1]))
+
+
+def due_scales(
+    *,
+    today: date,
+    enrolled: date,
+    last: dict[str, date],
+    route: Literal["specialist", "self_help"],
+    medication_changed: date | None = None,
+    after_crisis: date | None = None,
+    policy: RuntimePolicy | None = None,
+) -> tuple[str, ...]:
+    """Return each overdue form once. Therapy day zero is the recorded start/change date.
+
+    Due dates stay due until completion. Overlapping schedules choose the earlier
+    date; missed ticks do not create a burst of historical questionnaires.
+    """
+    p = policy or load_runtime()
+    due = set()
+    for scale in ("phq9", "pcl5"):
+        completed = last.get(scale)
+        dates = [(completed or enrolled) + timedelta(days=p.route_intervals[route][scale])]
+        if after_crisis and after_crisis <= today:
+            dates.append(
+                max(completed or after_crisis, after_crisis) + timedelta(days=p.route_intervals["after_crisis"][scale])
+            )
+        if medication_changed and medication_changed <= today:
+            for offset in p.medication_days:
+                milestone = medication_changed + timedelta(days=offset)
+                if completed is None or completed < milestone:
+                    dates.append(milestone)
+                    break
+            else:
+                dates.append(
+                    max(completed or medication_changed, medication_changed + timedelta(days=p.medication_days[-1]))
+                    + timedelta(days=p.medication_repeat_days)
+                )
+        if min(dates) <= today:
+            due.add(scale)
+    if (last.get("rus_audit_s") or enrolled) + timedelta(days=p.alcohol_repeat_days) <= today:
+        due.add("rus_audit_s")
+    return tuple(sorted(due))
+
+
+def diary_flags(history: dict[date, dict], today: date, policy: RuntimePolicy | None = None) -> tuple[str, ...]:
+    """Consecutive calendar dates; missing data never become zero or missed doses.
+
+    Baseline is the seven dates BEFORE the three-day deterioration window;
+    every baseline value is required. Nightmares OR short sleep must hold on
+    each of three nights. These candidate definitions require signed review.
+    """
+    p = policy or load_runtime()
+    flags = []
+    current = history.get(today, {})
+    if current.get("death_thoughts") == 1:
+        flags.append("suicide_assessment_now")
+    recent = [history.get(today - timedelta(days=i), {}) for i in range(p.mood_bad_days)]
+    baseline = [
+        history.get(today - timedelta(days=i), {}).get("mood")
+        for i in range(p.mood_bad_days, p.mood_bad_days + p.mood_baseline_days)
+    ]
+    if all(x is not None for x in baseline) and all(x.get("mood") is not None for x in recent):
+        mean = sum(baseline) / len(baseline)
+        if all(mean - x["mood"] >= p.mood_drop for x in recent):
+            flags.append("phq9_due_mood")
+    nights = [history.get(today - timedelta(days=i), {}) for i in range(p.sleep_bad_days)]
+    if all(
+        (r.get("nightmares") is not None and r["nightmares"] >= p.nightmares_at)
+        or (r.get("sleep_hours") is not None and r["sleep_hours"] < p.sleep_hours_below)
+        for r in nights
+    ):
+        flags.append("sleep_review")
+    doses = [history.get(today - timedelta(days=i), {}) for i in range(p.missed_medication_days)]
+    if all(r.get("medication") == "none" for r in doses):
+        flags.append("medication_contact")
+    if current.get("side_effects") == 1:
+        flags.append("side_effects_review")
+    if current.get("medication") == "partial":
+        flags.append("partial_medication_review")
+    return tuple(flags)

app/core/questionnaire.py

@@ -0,0 +1,337 @@
+"""Versioned intake rules for clinical review (#52).
+
+No patient wording or LLM calls. This module deliberately does not replace legacy
+cssrs_screen (whose q6 only covers three months) or reinterpret historical data.
+The dialog adapter must pass RAW answers, persist the policy version, and respect
+the returned status. Pending clinical policy never authorizes routine continuation.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime
+from pathlib import Path
+from typing import Literal
+
+import yaml
+from pydantic import BaseModel, ConfigDict, Field, model_validator
+
+from app.config import get_settings
+from app.core.scales import Sex, load_scale, score
+
+Risk = Literal["none", "low", "moderate", "high"]
+
+
+class Approval(BaseModel):
+    model_config = ConfigDict(extra="forbid", frozen=True, str_strip_whitespace=True)
+    reviewer: str = Field(min_length=1)
+    decision_reference: str = Field(min_length=1)
+    approved_at: datetime
+    policy_version: str = Field(min_length=1)
+
+    @model_validator(mode="after")
+    def _aware(self) -> Approval:
+        if self.approved_at.tzinfo is None:
+            raise ValueError("approval timestamp must include timezone")
+        return self
+
+
+class IntakePolicy(BaseModel):
+    model_config = ConfigDict(extra="forbid", frozen=True)
+    version: str = Field(min_length=1)
+    approval: Approval | None = None
+    cssrs_positive_policy: Literal["pending", "stop_any", "graded"] = "pending"
+    rus_audit_s_expand_at: int | None = Field(default=None, ge=1, le=12, strict=True)
+    rus_audit_s_expand_from_zone: int | None = Field(default=None, ge=1, le=4, strict=True)
+    phq2_expand_at: int = Field(ge=1, le=6, strict=True)
+    tsq_expand_at: int = Field(ge=1, le=10, strict=True)
+    phq9_psychologist_at: int = Field(ge=0, le=27, strict=True)
+    phq9_specialist_at: int = Field(ge=0, le=27, strict=True)
+    phq9_priority_at: int = Field(ge=0, le=27, strict=True)
+    pcl5_specialist_at: int = Field(ge=0, le=80, strict=True)
+    pcl5_priority_at: int = Field(ge=0, le=80, strict=True)
+    pcl5_behavior_flag_at: int = Field(ge=0, le=4, strict=True)
+
+    @model_validator(mode="after")
+    def _approval_matches(self) -> IntakePolicy:
+        if self.approval and (
+            self.approval.policy_version != self.version
+            or self.cssrs_positive_policy == "pending"
+            or (self.rus_audit_s_expand_from_zone is None and self.rus_audit_s_expand_at is None)
+        ):
+            raise ValueError("approval requires the matching version and resolved clinical decisions")
+        if self.rus_audit_s_expand_from_zone is not None and self.rus_audit_s_expand_at is not None:
+            raise ValueError("choose a total threshold OR a zone for alcohol expansion")
+        if not self.phq9_psychologist_at < self.phq9_specialist_at < self.phq9_priority_at:
+            raise ValueError("PHQ-9 thresholds must be ordered")
+        if self.pcl5_specialist_at >= self.pcl5_priority_at:
+            raise ValueError("PCL-5 thresholds must be ordered")
+        return self
+
+
+def load_policy(path: Path | None = None) -> IntakePolicy:
+    path = path or get_settings().config_dir / "questionnaire.yaml"
+    return IntakePolicy.model_validate(yaml.safe_load(path.read_text(encoding="utf-8")))
+
+
+@dataclass(frozen=True)
+class AlcoholResult:
+    scale: str
+    complete: bool
+    total: int | None
+    zone: int | None
+    missing: tuple[str, ...]
+    derived: tuple[str, ...] = ()
+
+
+def score_alcohol(answers: dict[str, int], *, short: bool, sex: Sex | None) -> AlcoholResult:
+    """RUS-AUDIT-S uses original items 3/9/10; unknown sex never borrows a cut-off."""
+    if sex not in (None, "m", "f"):
+        raise ValueError("sex must be m, f or None")
+    scale_id = "rus_audit_s" if short else "rus_audit"
+    # Validate BEFORE derivation: bools, floats and conflicting answers must not be hidden.
+    score(scale_id, answers, sex=sex)
+    values = dict(answers)
+    derived = []
+    if not short and values.get("rus_audit_1") == 0:
+        for item in ("rus_audit_2", "rus_audit_3"):
+            if values.get(item, 0) != 0:
+                raise ValueError("never drinks conflicts with positive consumption answers")
+            if item == "rus_audit_2" and item not in values:
+                values[item] = 0
+                derived.append(item)
+    result = score(scale_id, values, sex=sex)
+    missing = tuple(i for i in load_scale(scale_id).item_ids if i not in values)
+    zone = int(result.severity_code[-1]) if result.complete and sex is not None else None
+    return AlcoholResult(
+        scale_id, result.complete, result.total if result.complete else None, zone, missing, tuple(derived)
+    )
+
+
+def extend_alcohol(short_answers: dict[str, int], additional: dict[str, int]) -> dict[str, int]:
+    """Reuse original item IDs. Duplicate equal answers count once; conflicting ones fail."""
+    score("rus_audit_s", short_answers)
+    score("rus_audit", additional)
+    for key in short_answers.keys() & additional.keys():
+        if short_answers[key] != additional[key]:
+            raise ValueError(f"conflicting repeated answer: {key}")
+    return {**short_answers, **additional}
+
+
+def alcohol_expansion(result: AlcoholResult, policy: IntakePolicy) -> bool | None:
+    """None means unresolved, never a negative screen or permission to skip the full form."""
+    if result.scale != "rus_audit_s":
+        raise ValueError("expansion requires RUS-AUDIT-S")
+    if not policy.approval or not result.complete or result.zone is None:
+        return None
+    if policy.rus_audit_s_expand_at is not None:
+        return result.total >= policy.rus_audit_s_expand_at
+    assert policy.rus_audit_s_expand_from_zone is not None
+    return result.zone >= policy.rus_audit_s_expand_from_zone
+
+
+_CSSRS_ITEMS = ("q1", "q2", "q3", "q4", "q5", "q6_lifetime", "q6_recent")
+
+
+@dataclass(frozen=True)
+class CssrsResult:
+    complete: bool
+    risk: Risk | None
+    risk_lower_bound: Risk
+    positive: bool
+    required: tuple[str, ...]
+    missing: tuple[str, ...]
+    conflicts: tuple[str, ...]
+
+
+def score_cssrs_recent(answers: dict[str, int]) -> CssrsResult:
+    """Recent screener structure; q1-5 month, q6 lifetime + conditional three-month follow-up.
+
+    These keys describe data, NOT an official Russian translation. Positive signals
+    survive missing/contradictory answers; only complete consistent data yield final risk.
+    """
+    for key, value in answers.items():
+        if key not in _CSSRS_ITEMS or type(value) is not int or value not in (0, 1):
+            raise ValueError("C-SSRS requires known item IDs and integer 0/1 answers")
+    required = ["q1", "q2"]
+    if answers.get("q2") == 1:
+        required.extend(["q3", "q4", "q5"])
+    required.append("q6_lifetime")
+    if answers.get("q6_lifetime") == 1:
+        required.append("q6_recent")
+    conflicts = []
+    if answers.get("q2") == 0 and any(answers.get(k) == 1 for k in ("q3", "q4", "q5")):
+        conflicts.append("ideation_branch_conflict")
+    if answers.get("q6_lifetime") == 0 and answers.get("q6_recent") == 1:
+        conflicts.append("behavior_recency_conflict")
+    missing = tuple(k for k in required if k not in answers)
+    complete = not missing and not conflicts
+    lower: Risk = "none"
+    if any(answers.get(k) == 1 for k in ("q4", "q5", "q6_recent")):
+        lower = "high"
+    elif answers.get("q3") == 1 or answers.get("q6_lifetime") == 1:
+        lower = "moderate"
+    elif any(answers.get(k) == 1 for k in ("q1", "q2")):
+        lower = "low"
+    return CssrsResult(
+        complete, lower if complete else None, lower, 1 in answers.values(), tuple(required), missing, tuple(conflicts)
+    )
+
+
+@dataclass(frozen=True)
+class IntakeDecision:
+    policy_version: str
+    status: Literal["crisis", "clinical_review_required", "incomplete", "ready"]
+    primary_route: str | None
+    parallel_routes: tuple[str, ...]
+    flags: tuple[str, ...]
+    missing: tuple[str, ...]
+
+
+def _additional_flags(pcl5, alcohol, sex, traumatic_brain_injury, psychotropics, policy):
+    """Preserve independently observable flags even when the primary route is crisis.
+
+    Invalid unrelated data is marked for review rather than blocking a known crisis.
+    The routine branch still validates every answer strictly before routing.
+    """
+    flags = []
+    try:
+        score("pcl5", pcl5)
+        if any(pcl5.get(k, 0) >= policy.pcl5_behavior_flag_at for k in ("pcl5_15", "pcl5_16")):
+            flags.append("pcl5_behavior")
+    except ValueError:
+        flags.append("invalid_pcl5_answers")
+    try:
+        full = score_alcohol(alcohol, short=False, sex=sex)
+        short_values = {k: v for k, v in alcohol.items() if k in load_scale("rus_audit_s").item_ids}
+        short = score_alcohol(short_values, short=True, sex=sex)
+        zone = full.zone if full.complete else short.zone
+        if zone is not None and zone >= 2:
+            flags.append("alcohol_zone_ii_or_higher")
+        if psychotropics and (any(v > 0 for v in alcohol.values())):
+            flags.append("alcohol_psychotropics")
+        if full.zone == 4:
+            flags.append("alcohol_zone_iv")
+    except ValueError:
+        flags.append("invalid_alcohol_answers")
+    if traumatic_brain_injury:
+        flags.append("traumatic_brain_injury")
+    return flags
+
+
+def evaluate_intake(
+    *,
+    phq9: dict[str, int],
+    tsq: dict[str, int],
+    pcl5: dict[str, int],
+    alcohol: dict[str, int],
+    cssrs: dict[str, int],
+    sex: Sex | None,
+    trauma_exposure: bool | None,
+    help_now: bool = False,
+    traumatic_brain_injury: bool = False,
+    psychotropics: bool = False,
+    policy: IntakePolicy | None = None,
+) -> IntakeDecision:
+    """Two-stage candidate routing. No side effects, notifications, or claims of approval.
+
+    `alcohol` keeps full-scale IDs across both stages. Existing PHQ/PCL wording and
+    new C-SSRS content still need release review even with a signed routing policy.
+    """
+    policy = policy or load_policy()
+    if type(help_now) is not bool or type(traumatic_brain_injury) is not bool or type(psychotropics) is not bool:
+        raise ValueError("context flags must be bool")
+    if trauma_exposure is not None and type(trauma_exposure) is not bool:
+        raise ValueError("trauma_exposure must be bool or None")
+    # Independent flags survive the higher-priority crisis route.
+    flags = _additional_flags(pcl5, alcohol, sex, traumatic_brain_injury, psychotropics, policy)
+    # A manual emergency request does not depend on the completeness of any scale.
+    if help_now:
+        return IntakeDecision(policy.version, "crisis", "crisis", (), tuple(["help_now", *flags]), ())
+    # Act on explicit safety answers BEFORE processing unrelated, possibly invalid scales.
+    phq_item9 = phq9.get("phq9_9")
+    item9_positive = type(phq_item9) is int and phq_item9 in (1, 2, 3)
+    try:
+        c = score_cssrs_recent(cssrs)
+    except ValueError:
+        known_high = any(type(cssrs.get(k)) is int and cssrs[k] == 1 for k in ("q4", "q5", "q6_recent"))
+        if not item9_positive and not known_high:
+            raise
+        flags.extend(["invalid_cssrs_answers", "phq9_item9_positive" if item9_positive else "cssrs_positive"])
+        return IntakeDecision(policy.version, "crisis", "crisis", (), tuple(flags), ())
+    if item9_positive:
+        flags.append("phq9_item9_positive")
+        if c.complete and c.risk == "none":
+            flags.append("phq9_cssrs_discrepancy")
+    if c.positive:
+        flags.append("cssrs_positive")
+    flags.extend(c.conflicts)
+    if item9_positive or c.risk_lower_bound == "high":
+        return IntakeDecision(policy.version, "crisis", "crisis", (), tuple(flags), c.missing)
+    if c.positive and policy.approval and policy.cssrs_positive_policy == "stop_any":
+        return IntakeDecision(policy.version, "crisis", "crisis", (), tuple(flags), c.missing)
+    if c.positive and not policy.approval:
+        return IntakeDecision(policy.version, "clinical_review_required", None, (), tuple(flags), c.missing)
+
+    phq = score("phq9", phq9)
+    pc = score("tsq", tsq)
+    pcl = score("pcl5", pcl5)
+    full_alcohol = score_alcohol(alcohol, short=False, sex=sex)
+    short_values = {k: v for k, v in alcohol.items() if k in load_scale("rus_audit_s").item_ids}
+    short_alcohol = score_alcohol(short_values, short=True, sex=sex)
+    missing = list(c.missing)
+    phq2_items = ("phq9_1", "phq9_2")
+    missing.extend(k for k in phq2_items if k not in phq9)
+    phq2_complete = all(k in phq9 for k in phq2_items)
+    phq2_positive = phq2_complete and sum(phq9[k] for k in phq2_items) >= policy.phq2_expand_at
+    if phq2_positive:
+        missing.extend(k for k in load_scale("phq9").item_ids if k not in phq9)
+    if trauma_exposure is None:
+        missing.append("trauma_exposure")
+    if trauma_exposure is False and tsq:
+        flags.append("trauma_filter_conflict")
+    pc_positive = trauma_exposure is True and pc.complete and pc.total >= policy.tsq_expand_at
+    if trauma_exposure is True and not pc.complete:
+        missing.extend(k for k in load_scale("tsq").item_ids if k not in tsq)
+    if pc_positive and not pcl.complete:
+        missing.extend(k for k in load_scale("pcl5").item_ids if k not in pcl5)
+    # Partially administered optional scales also cannot be treated as completed negatives.
+    if len(phq9) > 2 and not phq.complete:
+        missing.extend(k for k in load_scale("phq9").item_ids if k not in phq9)
+    if pcl5 and not pcl.complete:
+        missing.extend(k for k in load_scale("pcl5").item_ids if k not in pcl5)
+    missing.extend(short_alcohol.missing)
+    if sex is None:
+        missing.append("alcohol_scoring_sex")
+    expand = alcohol_expansion(short_alcohol, policy)
+    if expand or (len(alcohol) > 3 and not full_alcohol.complete):
+        missing.extend(full_alcohol.missing)
+    if expand is None:
+        flags.append("alcohol_transition_unresolved")
+    if not policy.approval or c.conflicts or "trauma_filter_conflict" in flags:
+        return IntakeDecision(
+            policy.version, "clinical_review_required", None, (), tuple(flags), tuple(dict.fromkeys(missing))
+        )
+    if missing:
+        return IntakeDecision(policy.version, "incomplete", None, (), tuple(flags), tuple(dict.fromkeys(missing)))
+    # PHQ-2-negative does not invent a PHQ-9 total; it simply does not trigger expansion.
+    phq_total = phq.total if phq.complete else 0
+    pcl_total = pcl.total if pcl.complete else 0
+    if c.risk == "moderate":
+        primary = "urgent_psychiatrist"
+    elif phq_total >= policy.phq9_priority_at or pcl_total >= policy.pcl5_priority_at:
+        primary = "priority_psychiatrist"
+    elif phq_total >= policy.phq9_specialist_at or pcl_total >= policy.pcl5_specialist_at:
+        primary = "psychiatrist_or_psychotherapist"
+    elif phq_total >= policy.phq9_psychologist_at or pc_positive:
+        primary = "psychologist"
+    else:
+        primary = "self_help"
+    parallel = []
+    if full_alcohol.zone == 4:
+        parallel.append("narcologist")
+    if traumatic_brain_injury:
+        parallel.append("neurologist")
+    if c.risk == "low":
+        flags.append("cssrs_low_safety_plan")
+    return IntakeDecision(policy.version, "ready", primary, tuple(parallel), tuple(flags), ())

app/db/bootstrap.py

@@ -4,6 +4,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.clinical import ClinicalBase
 from app.db.identity import IdentityBase
 from app.db.session import clinical_engine, identity_engine

app/db/intake.py

@@ -0,0 +1,55 @@
+"""Versioned questionnaire records, without modifying historical scale results."""
+
+from datetime import date, datetime
+from typing import Any
+
+from sqlalchemy import Date, ForeignKey, Integer, String, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db.clinical import BigId, ClinicalBase
+from app.db.types import JSONType, TZDateTime
+
+
+class QuestionnaireRecord(ClinicalBase):
+    __tablename__ = "questionnaire_records"
+    id: Mapped[str] = mapped_column(String(32), primary_key=True)
+    token: Mapped[str] = mapped_column(String(32), index=True)
+    kind: Mapped[str] = mapped_column(String(32))
+    version: Mapped[str] = mapped_column(String(64))
+    policy_version: Mapped[str] = mapped_column(String(64))
+    started_at: Mapped[datetime] = mapped_column(TZDateTime)
+    completed_at: Mapped[datetime | None] = mapped_column(TZDateTime)
+    day: Mapped[date] = mapped_column(Date)
+    status: Mapped[str] = mapped_column(String(32))
+    answers: Mapped[dict[str, Any]] = mapped_column(JSONType, default=dict)
+    # Per-field time, recall period, source and derived values; totals may be null.
+    details: Mapped[dict[str, Any]] = mapped_column(JSONType, default=dict)
+
+
+class QuestionnaireNotice(ClinicalBase):
+    __tablename__ = "questionnaire_notices"
+    __table_args__ = (UniqueConstraint("token", "event_key"),)
+    id: Mapped[int] = mapped_column(BigId, primary_key=True, autoincrement=True)
+    token: Mapped[str] = mapped_column(String(32), index=True)
+    event_key: Mapped[str] = mapped_column(String(128))
+    alert_id: Mapped[int] = mapped_column(ForeignKey("alerts.id"))
+    created_at: Mapped[datetime] = mapped_column(TZDateTime)
+    attempted_at: Mapped[datetime | None] = mapped_column(TZDateTime)
+    accepted_at: Mapped[datetime | None] = mapped_column(TZDateTime)
+    attempts: Mapped[int] = mapped_column(Integer, default=0)
+    status: Mapped[str] = mapped_column(String(32), default="pending")
+    error_code: Mapped[str | None] = mapped_column(String(80))
+
+
+class QuestionnaireTask(ClinicalBase):
+    __tablename__ = "questionnaire_tasks"
+    __table_args__ = (UniqueConstraint("token", "task_key"),)
+    id: Mapped[int] = mapped_column(BigId, primary_key=True, autoincrement=True)
+    token: Mapped[str] = mapped_column(String(32), index=True)
+    task_key: Mapped[str] = mapped_column(String(128))
+    kind: Mapped[str] = mapped_column(String(32))
+    due_at: Mapped[datetime] = mapped_column(TZDateTime)
+    status: Mapped[str] = mapped_column(String(32), default="pending")
+    record_id: Mapped[str | None] = mapped_column(String(32))
+    delivery: Mapped[list] = mapped_column(JSONType, default=list)
+    sent_at: Mapped[datetime | None] = mapped_column(TZDateTime)

app/dialog/crisis.py

@@ -4,7 +4,7 @@ Decisions:
   * The crisis script (`app.core.safety.crisis_reply`, `config/texts/crisis.yaml`) is sent verbatim, once per
     open crisis. "Open" = `data["crisis"]["open"]` — set when the crisis starts, cleared only by the doctor's
     acknowledgement (`doctor_ack`), so a patient repeating the phrase does not get the same long script again:
-    they get a short fixed "я рядом, врач уже уведомлён" + emergency resources instead.
+    they get a short fixed support message and emergency resources instead.
   * Entering the crisis drops unfinished scale sessions (their answered items are already stored item by item):
     nothing else is asked while the patient is in crisis, except the C-SSRS screen.
   * Without valid consent (before consent, after refusal/revocation, minors) we still answer a crisis message —

app/dialog/engine.py

@@ -22,15 +22,18 @@ Pipeline for one incoming event (SPEC §3, guardrails #1, #2, #3):
 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 typing import Any
+from weakref import WeakValueDictionary
 
 from sqlalchemy.exc import InterfaceError, OperationalError
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core import clock, safety
+from app.core.intake_runtime import intake_enabled
 from app.core.privacy import scrub
 from app.core.safety import SafetyResult, crisis_reply
 from app.core.types import RiskLevel, ScrubbedText
@@ -45,12 +48,14 @@ 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.texts import DialogTexts, load_dialog_texts
 from app.llm.gateway import LLMGateway, get_gateway
+from app.services import questionnaire as intake_service
 from app.services.contracts import MonitoringService, get_monitoring
 
 log = logging.getLogger(__name__)
 
 # Enter hooks may chain (scale done → next onboarding state → next scale); a cycle must not hang a request.
 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)
 
@@ -76,8 +81,13 @@ class Engine:
     async def handle(self, token: str, event: Event) -> list[Reply]:
         texts = load_dialog_texts()
         try:
-            async with get_clinical_session() as session:
-                return await self._handle(token, event, session, texts)
+            lock = _patient_locks.setdefault(token, asyncio.Lock())
+            async with lock:
+                async with get_clinical_session() as session:
+                    await intake_service.lock_patient(session, token)
+                    replies = await self._handle(token, event, session, texts)
+                await intake_service.dispatch_notices(token)
+                return replies
         except _DB_DOWN as exc:
             log.error("dialog db unavailable token=%s event=%s error=%s", token, event.type, type(exc).__name__)
             return _degraded(event, texts)
@@ -117,6 +127,13 @@ class Engine:
                     session, token=token, direction="out", text=scrub(reply.text), ts=ctx.now, fsm_state=ctx.state
                 )
         await self._save_state(ctx)
+        task_id = event.params.get("questionnaire_task")
+        if event.type == "system" and task_id and run.replies:
+            from app.db.intake import QuestionnaireTask
+
+            task = await session.get(QuestionnaireTask, task_id)
+            if task and task.token == token:
+                task.delivery = [r.model_dump() for r in run.replies]
         log.info(
             "dialog token=%s event=%s code=%s state=%s->%s replies=%d",
             token,
@@ -143,6 +160,12 @@ class Engine:
         if event.type == "button":
             if consented and event.button == "consent:revoke":
                 return await revoke(ctx)
+            if consented and event.button == "intake:help":
+                from app.db.intake import QuestionnaireRecord
+                from app.dialog.flows.intake import urgent
+
+                row = await ctx.session.get(QuestionnaireRecord, ctx.data.get("questionnaire_record", ""))
+                return await urgent(ctx, row)
             if consented:
                 await self._store_in(ctx, scrub(f"[button] {event.button}"))  # button data comes from the client
             return await self._dispatch(ctx, event)
@@ -153,6 +176,14 @@ class Engine:
             if not consented:
                 return Step()  # nothing is pushed to people without consent
             if event.system == "scale_due" and event.params.get("scale"):
+                if intake_enabled() and ctx.data.get("adult") is True:
+                    scale = str(event.params["scale"])
+                    kind = "cssrs_since_contact" if scale in ("cssrs_screen", "cssrs") else scale
+                    if kind in ("phq9", "pcl5", "rus_audit_s", "cssrs_since_contact"):
+                        await intake_service.queue_task(
+                            ctx.session, ctx.token, kind, f"external:{kind}:{ctx.now.date()}", ctx.now
+                        )
+                    return Step()
                 scheduled = bool(event.params.get("scheduled", True))
                 pending = scale_session.add_pending(ctx.data, str(event.params["scale"]), scheduled)
                 return Step(data={"pending_scales": pending})  # starts now if resting, else after the current step
@@ -181,11 +212,29 @@ class Engine:
         if result.level == RiskLevel.ACUTE:
             handled = "crisis_repeat" if crisis_open(ctx.data) else "crisis_script"
             log.warning("safety acute token=%s rules=%s handled=%s", ctx.token, result.matched_rules, handled)
-            step = open_crisis(ctx, result.categories, source=f"safety:{result.source}")
+            if ctx.data.get("questionnaire_record") or (intake_enabled() and ctx.data.get("adult") is True):
+                from app.db.intake import QuestionnaireRecord
+                from app.dialog.flows.intake import urgent
+
+                row = await ctx.session.get(QuestionnaireRecord, ctx.data.get("questionnaire_record", ""))
+                step = await urgent(ctx, row, code="safety_acute", categories=result.categories)
+            else:
+                step = open_crisis(ctx, result.categories, source=f"safety:{result.source}")
             step.data["safety_events"] = self._safety_events(ctx, result, handled)
             step.notify_monitoring = True  # lane B escalates to the doctor on call
             return step
 
+        if result.level == RiskLevel.POSSIBLE and (
+            ctx.data.get("questionnaire_record") or (intake_enabled() and ctx.data.get("adult") is True)
+        ):
+            from app.db.intake import QuestionnaireRecord
+            from app.dialog.flows.intake import manual_review
+
+            row = await ctx.session.get(QuestionnaireRecord, ctx.data.get("questionnaire_record", ""))
+            if row is None:
+                row = await intake_service.create_record(ctx, "cssrs_since_contact")
+            return await manual_review(ctx, row, "cssrs_since_contact_required")
+
         if result.level == RiskLevel.POSSIBLE:
             screen = ctx.texts.settings.risk_screen_scale
             if self._can_screen(ctx, screen):
@@ -252,6 +301,8 @@ class Engine:
 
     async def _start_pending(self, ctx: Ctx) -> Step:
         """Start a queued scale when nothing else is being asked. In crisis only the risk screen may run."""
+        if intake_enabled():
+            return Step()
         pending = list(ctx.data.get("pending_scales") or [])
         if not pending or ctx.state not in RESTING_STATES:
             return Step()

app/dialog/flows/core.py

@@ -42,6 +42,20 @@ def menu_reply(ctx: Ctx, text: str | None = None) -> Reply:
         [Button(label=b.contact_doctor, data="menu:contact_doctor")],
         [Button(label=b.revoke, data="consent:revoke")],
     ]
+    from app.core.intake_questions import texts
+    from app.core.intake_runtime import intake_enabled
+
+    if intake_enabled():
+        buttons.insert(0, [Button(label=texts()["help"], data="intake:help")])
+        buttons.insert(
+            0,
+            [
+                Button(
+                    label=texts()["resume"] if ctx.data.get("questionnaire_record") else texts()["start"],
+                    data="intake:resume" if ctx.data.get("questionnaire_record") else "intake:start",
+                )
+            ],
+        )
     return Reply(text=text or ctx.texts.idle.menu_prompt, buttons=buttons)
 
 
@@ -53,6 +67,10 @@ async def onboarding_entry(ctx: Ctx, after: str | None = None) -> str:
     profile (after `after`, if given) that has such a state registered; `"idle"` when none is left."""
     profile = await ctx.profile()
     modules = list(profile.modules) if profile is not None and profile.modules else []
+    from app.core.intake_runtime import intake_enabled
+
+    if after is None and intake_enabled() and any(m in modules for m in ("veteran", "depression")):
+        return "intake.start"
     if after in modules:
         modules = modules[modules.index(after) + 1 :]
     for module in modules:
@@ -275,6 +293,8 @@ async def _reach_out(ctx: Ctx, event: Event) -> Step:
 
 def close_crisis(ctx: Ctx) -> Step:
     """Doctor acknowledged the crisis: clear the flag; a C-SSRS still running returns to idle afterwards."""
+    if not (ctx.data.get("crisis") or {}).get("open") and ctx.state != "crisis":
+        return Step()
     crisis = {**(ctx.data.get("crisis") or {}), "open": False, "closed_at": ctx.now.isoformat()}
     scales = {
         sid: {**s, "return_state": "idle"} if s.get("return_state") == "crisis" else s
@@ -311,7 +331,11 @@ async def revoke(ctx: Ctx) -> Step:
     """
     async with get_identity_session() as s:
         await identity_repo.revoke_consent(s, token=ctx.token, now=ctx.now)
+    from app.services.questionnaire import cancel_pending
+
+    await cancel_pending(ctx.session, ctx.token)
     data = {
+        "questionnaire_record": None,
         "consent": False,
         "revoked_at": ctx.now.isoformat(),
         "delete_requested": True,

app/dialog/flows/intake.py

@@ -0,0 +1,376 @@
+"""Versioned, resumable questionnaires. Disabled until the clinical release gate opens."""
+
+import math
+from dataclasses import asdict, replace
+from zoneinfo import ZoneInfo
+
+from app.core.intake_questions import (
+    content,
+    cssrs_answers,
+    diary_questions,
+    group_answers,
+    next_intake,
+    scale_question,
+    texts,
+)
+from app.core.intake_runtime import diary_flags, intake_enabled, load_runtime
+from app.core.questionnaire import alcohol_expansion, load_policy, score_alcohol, score_cssrs_recent
+from app.core.scales import load_scale
+from app.core.types import AlertLevel
+from app.db.intake import QuestionnaireRecord, QuestionnaireTask
+from app.dialog.contracts import Button, Event, Reply
+from app.dialog.crisis import crisis_open, open_crisis
+from app.dialog.fsm import Step, on_enter, state, trigger
+from app.services import questionnaire as service
+
+
+def allowed(ctx):
+    return ctx.data.get("consent") is True and ctx.data.get("adult") is True and intake_enabled()
+
+
+def next_question(row):
+    a = row.answers
+    if row.kind == "intake":
+        return next_intake(a, load_policy(), row.details.get("sex"))
+    if row.kind == "diary":
+        for q in diary_questions(
+            row.details["modules"], suicide_monitoring=row.details["suicide_monitoring"], weekday=row.day.weekday()
+        ):
+            if q.id in a:
+                continue
+            if q.id == "medication_reason" and a.get("medication") not in ("partial", "none"):
+                continue
+            if q.id == "side_effects" and a.get("medication") == "not_prescribed":
+                continue
+            if q.id == "thought_intensity" and a.get("death_thoughts") != 1:
+                continue
+            return q
+        return None
+    if row.kind in ("phq9", "pcl5", "rus_audit_s"):
+        sid = row.kind
+        if sid == "rus_audit_s":
+            short = score_alcohol(
+                {k: a[k] for k in load_scale(sid).item_ids if k in a}, short=True, sex=row.details.get("sex")
+            )
+            if alcohol_expansion(short, load_policy()):
+                sid = "rus_audit"
+        for item in load_scale(sid).item_ids:
+            if item == "rus_audit_2" and a.get("rus_audit_1") == 0:
+                continue
+            if item not in a:
+                q = scale_question(sid, item)
+                if sid == "pcl5" and row.details["weekly_pcl"]:
+                    q = replace(q, text=q.text.replace("последний месяц", "последнюю неделю"), period="past_week")
+                return q
+        if row.details.get("medication_active") and "side_effects" not in a:
+            return next(
+                q
+                for q in diary_questions([], suicide_monitoring=False, weekday=row.day.weekday())
+                if q.id == "side_effects"
+            )
+    return None
+
+
+def question_reply(row, q):
+    t = texts()
+    # Nonce + item + option index rejects old/replayed callbacks without exposing raw answers.
+    prefix = f"iq:{row.id[:12]}:{q.id}:"
+    buttons = [[Button(label=label, data=prefix + str(i))] for i, (_, label) in enumerate(q.options)]
+    if q.optional:
+        buttons.append([Button(label=t["comment_skip"], data=prefix + "skip")])
+    buttons.extend([[Button(label=t["help"], data="intake:help")], [Button(label=t["pause"], data="intake:pause")]])
+    return Reply(text=q.text + "\n\n" + t["as_of"].format(day=row.day.isoformat()), buttons=buttons, kind="question")
+
+
+async def urgent(ctx, row=None, *, code="help_now", categories=None):
+    if crisis_open(ctx.data):
+        return open_crisis(ctx, categories or [], source="questionnaire:" + code)
+    profile = await ctx.profile()
+    if profile:
+        profile.enhanced_monitoring = True
+    flags = list(service.decision(row).flags) if row else []
+    if row and (row.answers.get("death_thoughts") == 1 or row.answers.get("phq9_9", 0) > 0):
+        flags.append("cssrs_since_contact_required")
+    if row:
+        await service.finish(ctx, row, "interrupted", extra={"stop_reason": code, "flags": flags})
+    await service.alert(
+        ctx.session,
+        ctx.token,
+        f"{row.id if row else ctx.now.isoformat()}:{code}",
+        code,
+        ctx.now,
+        level=AlertLevel.ACUTE,
+        params={"record_id": row.id if row else None, "flags": flags},
+    )
+    step = open_crisis(ctx, categories or [], source="questionnaire:" + code)
+    step.data["questionnaire_record"] = None
+    step.data["intake_review_required"] = True
+    for reply in step.replies:
+        reply.buttons.append([Button(label=ctx.texts.safety_plan.title.rstrip(":"), data="safety_plan")])
+    return step
+
+
+async def manual_review(ctx, row, code):
+    await service.finish(ctx, row, "review_required", extra={"stop_reason": code})
+    await service.alert(
+        ctx.session, ctx.token, f"{row.id}:{code}", code, ctx.now, level=AlertLevel.HIGH, params={"record_id": row.id}
+    )
+    return Step(
+        replies=[
+            Reply(
+                text=texts()["missing_followup"] if code == "cssrs_since_contact_required" else texts()["review"],
+                buttons=[[Button(label=texts()["help"], data="intake:help")]],
+                kind="info",
+            )
+        ],
+        next_state="idle",
+        data={"questionnaire_record": None, "intake_review_required": True},
+    )
+
+
+async def advance(ctx, row):
+    a = row.answers
+    c = score_cssrs_recent(cssrs_answers(a))
+    policy = load_policy()
+    if a.get("phq9_9", 0) > 0 or a.get("death_thoughts") == 1 or c.risk_lower_bound == "high":
+        return await urgent(ctx, row, code="suicide_signal")
+    if c.positive and policy.cssrs_positive_policy == "stop_any":
+        return await urgent(ctx, row, code="cssrs_positive")
+    if c.conflicts:
+        return await manual_review(ctx, row, "cssrs_conflict")
+    # Finish the risk questions before any routine question; a moderate result stops here.
+    if c.complete and c.risk == "moderate":
+        return await manual_review(ctx, row, "urgent_psychiatrist")
+    if c.complete and c.risk == "low":
+        profile = await ctx.profile()
+        if profile:
+            profile.enhanced_monitoring = True
+        await service.alert(
+            ctx.session,
+            ctx.token,
+            row.id + ":cssrs_low",
+            "cssrs_low_safety_plan",
+            ctx.now,
+            params={"record_id": row.id},
+        )
+    if row.kind == "cssrs_since_contact":
+        # Recent and Since Last Contact are different instruments. No silent substitution.
+        return await manual_review(ctx, row, "cssrs_since_contact_required")
+    try:
+        q = next_question(row)
+        # Validate preserved short answers against a newly answered full-form item 1.
+        score_alcohol(group_answers(a, "rus_audit"), short=False, sex=row.details.get("sex"))
+    except ValueError:
+        return await manual_review(ctx, row, "answer_conflict")
+    if q:
+        replies = [question_reply(row, q)]
+        if c.complete and c.risk == "low" and not row.details.get("plan_shown"):
+            from app.dialog.actions import safety_plan_reply
+
+            replies.insert(0, await safety_plan_reply(ctx))
+            row.details = {**row.details, "plan_shown": True}
+        return Step(replies=replies, next_state="intake.collect", data={"questionnaire_record": row.id})
+    extra = {}
+    if a.get("side_effects") == 1:
+        await service.alert(
+            ctx.session,
+            ctx.token,
+            row.id + ":side_effects_review",
+            "side_effects_review",
+            ctx.now,
+            params={"record_id": row.id},
+        )
+    if row.kind == "intake":
+        result = service.decision(row)
+        if result.status != "ready":
+            return await manual_review(ctx, row, "intake_incomplete_review")
+        extra = {"decision": asdict(result), "totals": service.summary(row)}
+        if a.get("recent_trauma") == 1:
+            extra["recent_trauma"] = True
+        if result.primary_route != "self_help" or result.flags or result.parallel_routes or a.get("recent_trauma"):
+            await service.alert(
+                ctx.session,
+                ctx.token,
+                row.id + ":route",
+                "intake_route_review",
+                ctx.now,
+                params={"record_id": row.id, **extra},
+            )
+    elif row.kind == "diary":
+        history = {
+            r.day: r.answers
+            for r in await service.records(ctx.session, ctx.token)
+            if r.kind == "diary" and r.status == "complete"
+        }
+        history[row.day] = a
+        flags = diary_flags(history, row.day)
+        extra["flags"] = flags
+        for flag in flags:
+            await service.alert(
+                ctx.session, ctx.token, row.id + ":" + flag, flag, ctx.now, params={"record_id": row.id}
+            )
+            if flag == "phq9_due_mood":
+                await service.queue_task(ctx.session, ctx.token, "phq9", row.id + ":phq9", ctx.now)
+    else:
+        extra["totals"] = service.summary(row)
+        if row.kind in ("phq9", "pcl5"):
+            sid = row.kind
+            total = extra["totals"][sid]["total"]
+            priority = getattr(policy, sid + "_priority_at")
+            specialist = getattr(policy, sid + "_specialist_at")
+            extra["route"] = (
+                "priority_psychiatrist"
+                if total >= priority
+                else ("psychiatrist_or_psychotherapist" if total >= specialist else "review_with_treating_clinician")
+            )
+            prior = next(
+                (
+                    r
+                    for r in reversed(await service.records(ctx.session, ctx.token))
+                    if r.id != row.id
+                    and r.status == "complete"
+                    and r.version == row.version
+                    and r.details.get("totals", {}).get(sid, {}).get("complete")
+                    and (sid != "pcl5" or r.details.get("weekly_pcl", False) == row.details.get("weekly_pcl", False))
+                ),
+                None,
+            )
+            if prior:
+                change = total - prior.details["totals"][sid]["total"]
+                extra["delta"] = {
+                    "baseline_record": prior.id,
+                    "value": change,
+                    "retriage": change >= getattr(load_runtime(), sid + "_retriage_delta"),
+                }
+        if row.kind == "rus_audit_s" and not row.details.get("sex"):
+            return await manual_review(ctx, row, "alcohol_sex_unknown")
+        if extra["totals"]["rus_audit"].get("zone") == 4:
+            extra["parallel_routes"] = ["narcologist"]
+        retriage = extra.get("delta", {}).get("retriage", False)
+        await service.alert(
+            ctx.session,
+            ctx.token,
+            row.id + ":followup",
+            "followup_retriage_required" if retriage else "followup_review",
+            ctx.now,
+            level=AlertLevel.HIGH if retriage or extra.get("route") == "priority_psychiatrist" else AlertLevel.MODERATE,
+            params={"record_id": row.id, **extra},
+        )
+    await service.finish(ctx, row, "complete", extra=extra)
+    from app.dialog.flows.core import menu_reply
+
+    replies = [menu_reply(ctx, texts()["done"])]
+    alcohol_total = extra.get("totals", {}).get("rus_audit", {})
+    if alcohol_total.get("zone") in (2, 3):
+        replies.insert(0, Reply(text=texts()["alcohol_advice"], kind="info"))
+    return Step(replies=replies, next_state="idle", data={"questionnaire_record": None})
+
+
+async def start(ctx, kind="intake", task=None):
+    if not allowed(ctx) or crisis_open(ctx.data):
+        if task:
+            return Step()
+        return Step(replies=[Reply(text=texts()["unavailable"], kind="info")])
+    existing = await service.records(ctx.session, ctx.token)
+    active = next((r for r in reversed(existing) if r.status in ("active", "paused")), None)
+    if active:
+        if task and task.record_id != active.id:
+            return Step()
+        if (
+            active.policy_version != load_policy().version
+            or active.version != content()["version"]
+            or active.details.get("runtime_version") != load_runtime().version
+        ):
+            return await manual_review(ctx, active, "questionnaire_version_changed")
+        active.status = "active"
+        return await advance(ctx, active)
+    if ctx.data.get("intake_review_required") and kind != "cssrs_since_contact":
+        if task:
+            return Step()
+        return Step(replies=[Reply(text=texts()["review"], kind="info")])
+    row = await service.create_record(ctx, kind, task=task)
+    return await advance(ctx, row)
+
+
+@on_enter("intake.start")
+async def enter(ctx):
+    return await start(ctx)
+
+
+@trigger("button:intake:start")
+@trigger("button:intake:resume")
+async def start_button(ctx, event):
+    return await start(ctx)
+
+
+@trigger("button:menu:checkin", replace=True)
+@trigger("system:daily_checkin", replace=True)
+async def checkin(ctx, event):
+    if not intake_enabled():
+        return Step(replies=[Reply(text=ctx.texts.idle.stub_checkin)])
+    if event.params.get("questionnaire_task"):
+        task = await ctx.session.get(QuestionnaireTask, event.params["questionnaire_task"])
+        if task is None or task.token != ctx.token or task.status not in ("pending", "active"):
+            return Step()
+        return await start(ctx, task.kind, task)
+    existing = await service.records(ctx.session, ctx.token)
+    if not any(r.kind == "intake" and r.status == "complete" for r in existing):
+        return await start(ctx)
+    day = ctx.now.astimezone(ZoneInfo(load_runtime().timezone)).date()
+    if any(r.kind == "diary" and r.day == day and r.status == "complete" for r in existing):
+        return Step(replies=[Reply(text=texts()["done"], kind="info")])
+    risk_task = await service.queue_task(ctx.session, ctx.token, "cssrs_since_contact", f"cssrs:{day}", ctx.now)
+    if risk_task.status in ("pending", "active"):
+        await service.queue_task(ctx.session, ctx.token, "diary", f"diary:{day}", ctx.now)
+        return await start(ctx, "cssrs_since_contact", risk_task)
+    return await start(ctx, "diary")
+
+
+@state("intake")
+async def collect(ctx, event: Event):
+    if not allowed(ctx):
+        return Step(replies=[Reply(text=texts()["unavailable"], kind="info")], next_state="idle")
+    row = await ctx.session.get(QuestionnaireRecord, ctx.data.get("questionnaire_record", ""))
+    if row is None or row.token != ctx.token or row.status != "active":
+        return Step(next_state="idle")
+    if (
+        row.policy_version != load_policy().version
+        or row.version != content()["version"]
+        or row.details.get("runtime_version") != load_runtime().version
+    ):
+        return await manual_review(ctx, row, "questionnaire_version_changed")
+    if event.type == "system":
+        task_id = event.params.get("questionnaire_task")
+        if task_id and row.details.get("task_id") == task_id:
+            return await advance(ctx, row)
+        return Step()
+    if event.button == "intake:pause":
+        row.status = "paused"
+        return Step(
+            replies=[Reply(text=texts()["paused"], buttons=[[Button(label=texts()["resume"], data="intake:resume")]])],
+            next_state="idle",
+        )
+    q = next_question(row)
+    if q is None:
+        return await advance(ctx, row)
+    prefix = f"iq:{row.id[:12]}:{q.id}:"
+    value = None
+    valid = False
+    if event.type == "button" and (event.button or "").startswith(prefix):
+        index = event.button[len(prefix) :]
+        if index == "skip" and q.optional:
+            valid = True
+        elif index.isdigit() and int(index) < len(q.options):
+            value, valid = q.options[int(index)][0], True
+    elif event.type == "text" and q.numeric:
+        try:
+            value = float(str(ctx.text).replace(",", "."))
+            valid = math.isfinite(value) and 0 <= value <= 24
+        except ValueError:
+            pass
+    elif event.type == "text" and q.optional:
+        value, valid = str(ctx.text)[:2000], True
+    if not valid:
+        return Step(replies=[Reply(text=texts()["numeric" if q.numeric else "invalid"]), question_reply(row, q)])
+    service.save_answer(row, q, value, ctx.now)
+    return await advance(ctx, row)

app/scheduler/jobs.py

@@ -1,13 +1,176 @@
-"""Scheduler on the (virtual) demo clock — stub until #13. Called from app.main lifespan."""
+"""Questionnaire jobs on the demo clock; clinical release disabled by default.
+
+One scheduler per database. PostgreSQL's advisory transaction lock serializes
+workers; an asyncio lock does the same within a SQLite demonstration process.
+Deliveries are at least once: the channel protocol offers no idempotency key.
+"""
+
+import asyncio
+import logging
+from contextlib import suppress
+from datetime import datetime
+from zoneinfo import ZoneInfo
+
+from sqlalchemy import select, text
+
+from app.core import clock
+from app.core.intake_runtime import due_scales, intake_enabled, load_runtime
+from app.db import repo
+from app.db.clinical import PatientProfile
+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.services import questionnaire as service
+
+log = logging.getLogger(__name__)
+_worker = None
+_lock = asyncio.Lock()
 
 
-async def start() -> None:
-    """Start periodic ticking (#13)."""
+async def plan():
+    now = clock.now()
+    p = load_runtime()
+    local = now.astimezone(ZoneInfo(p.timezone))
+    async with get_clinical_session() as session:
+        profiles = (await session.scalars(select(PatientProfile))).all()
+        for profile in profiles:
+            state = await repo.get_dialog_state(session, profile.token)
+            if not state or state.data.get("consent") is not True or state.data.get("adult") is not True:
+                continue
+            rows = await service.records(session, profile.token)
+            initial = next((r for r in reversed(rows) if r.kind == "intake" and r.status == "complete"), None)
+            if initial is None:
+                continue
+            last_activity = max([r.day for r in rows if r.kind == "diary" and r.status == "complete"] + [initial.day])
+            high_risk = profile.enhanced_monitoring or bool(state.data.get("crisis"))
+            if high_risk and (local.date() - last_activity).days >= p.high_risk_silence_days:
+                await service.alert(session, profile.token, f"silence:{last_activity}", "high_risk_silence", now)
+            if (state.data.get("crisis") or {}).get("open") or state.data.get("intake_review_required"):
+                continue
+            if local.hour >= p.daily_hour and not any(
+                r.kind == "diary" and r.day == local.date() and r.status == "complete" for r in rows
+            ):
+                await service.queue_task(session, profile.token, "diary", f"diary:{local.date()}", now)
+            last = {}
+            for row in rows:
+                if row.status != "complete":
+                    continue
+                for sid in ("phq9", "pcl5", "rus_audit_s"):
+                    if row.details.get("totals", {}).get(sid, {}).get("complete"):
+                        last[sid] = row.completed_at.astimezone(ZoneInfo(p.timezone)).date()
+            route = initial.details["decision"]["primary_route"]
+            prescriptions = await integrations.prescriptions.get_prescriptions(profile.token)
+            changes = [
+                v.start_date
+                for v in prescriptions
+                if v.start_date and v.start_date <= local.date() and (v.end_date is None or v.end_date >= local.date())
+            ]
+            if profile.therapy_start and profile.therapy_start <= local.date():
+                changes.append(profile.therapy_start)
+            closed = (state.data.get("crisis") or {}).get("closed_at")
+            crisis_date = datetime.fromisoformat(closed).astimezone(ZoneInfo(p.timezone)).date() if closed else None
+            scales = due_scales(
+                today=local.date(),
+                enrolled=initial.day,
+                last=last,
+                route="self_help" if route == "self_help" else "specialist",
+                medication_changed=max(changes) if changes else None,
+                after_crisis=crisis_date,
+                policy=p,
+            )
+            if scales or local.hour >= p.daily_hour:
+                await service.queue_task(session, profile.token, "cssrs_since_contact", f"cssrs:{local.date()}", now)
+            for sid in scales:
+                if sid == "pcl5" and initial.answers.get("trauma") != 1:
+                    continue
+                await service.queue_task(session, profile.token, sid, f"{sid}:{local.date()}", now)
 
 
-async def stop() -> None:
-    """Stop the scheduler (#13)."""
+async def deliver():
+    async with get_clinical_session() as session:
+        task_ids = list(
+            (
+                await session.scalars(
+                    select(QuestionnaireTask.id)
+                    .where(
+                        QuestionnaireTask.sent_at.is_(None),
+                        QuestionnaireTask.status != "cancelled",
+                        QuestionnaireTask.due_at <= clock.now(),
+                    )
+                    .order_by((QuestionnaireTask.kind != "cssrs_since_contact"), QuestionnaireTask.id)
+                )
+            ).all()
+        )
+    for task_id in task_ids:
+        async with get_clinical_session() as session:
+            task = await session.get(QuestionnaireTask, task_id)
+            state = await repo.get_dialog_state(session, task.token)
+            if not state or state.data.get("consent") is not True or state.data.get("adult") is not True:
+                task.status, task.delivery = "cancelled", []
+                continue
+            if (state.data.get("crisis") or {}).get("open"):
+                continue
+            if task.status == "done" and not task.delivery:
+                continue
+            token = task.token
+            payload = task.delivery
+            if not payload and state.state not in ("idle", "intake.collect"):
+                continue
+        if not payload:
+            await get_engine().handle(
+                token, Event(type="system", system="daily_checkin", params={"questionnaire_task": task_id})
+            )
+        # Serialize delivery with withdrawal of consent and user answers.
+        async with get_clinical_session() as session:
+            await service.lock_patient(session, token)
+            state = await repo.get_dialog_state(session, token)
+            task = await session.get(QuestionnaireTask, task_id)
+            if not state or state.data.get("consent") is not True or not task.delivery:
+                continue
+            try:
+                await asyncio.wait_for(
+                    integrations.outbox.push(token, [Reply.model_validate(r) for r in task.delivery]), timeout=10
+                )
+            except Exception as exc:
+                log.warning("questionnaire outbox error=%s", type(exc).__name__)
+            else:
+                task.sent_at = clock.now()
 
 
-async def run_until_now() -> None:
-    """Replay all due events up to `app.core.clock.now()` in order, idempotently (#13)."""
+async def run_until_now():
+    async with _lock, get_clinical_session() as guard:
+        if guard.bind.dialect.name == "postgresql" and not await guard.scalar(
+            text("SELECT pg_try_advisory_xact_lock(520053)")
+        ):
+            return
+        await service.dispatch_notices()
+        if not intake_enabled():
+            return
+        await plan()
+        await deliver()
+        await service.dispatch_notices()
+
+
+async def _run():
+    while True:
+        try:
+            await run_until_now()
+        except Exception as exc:
+            log.error("questionnaire scheduler error=%s", type(exc).__name__)
+        await asyncio.sleep(30)
+
+
+async def start():
+    global _worker
+    if _worker is None and intake_enabled():
+        _worker = asyncio.create_task(_run())
+
+
+async def stop():
+    global _worker
+    if _worker is not None:
+        _worker.cancel()
+        with suppress(asyncio.CancelledError):
+            await _worker
+        _worker = None

app/services/questionnaire.py

@@ -0,0 +1,264 @@
+"""Transactional questionnaire storage and a durable doctor notification queue.
+
+Notifier acceptance is not delivery or a doctor's acknowledgement. At-least-once
+retries can duplicate delivery if a process dies after sending but before commit.
+"""
+
+import asyncio
+from dataclasses import asdict
+from datetime import timedelta
+from uuid import uuid4
+from zoneinfo import ZoneInfo
+
+from sqlalchemy import select
+
+from app.core import clock
+from app.core.intake_questions import content, cssrs_answers, group_answers
+from app.core.intake_runtime import load_runtime, month_before
+from app.core.questionnaire import evaluate_intake, load_policy, score_alcohol, score_cssrs_recent
+from app.core.scales import load_scale, score
+from app.core.types import AlertLevel
+from app.db import repo
+from app.db.clinical import Alert, DialogState
+from app.db.intake import QuestionnaireNotice, QuestionnaireRecord, QuestionnaireTask
+from app.db.session import get_clinical_session
+from app.integrations.base import MemoryNotifier, integrations
+
+
+async def records(session, token):
+    return list(
+        (
+            await session.scalars(
+                select(QuestionnaireRecord)
+                .where(QuestionnaireRecord.token == token)
+                .order_by(QuestionnaireRecord.started_at, QuestionnaireRecord.id)
+            )
+        ).all()
+    )
+
+
+async def create_record(ctx, kind, *, task=None):
+    profile = await ctx.profile()
+    runtime = load_runtime()
+    prior = next(
+        (r for r in reversed(await records(ctx.session, ctx.token)) if r.kind == "intake" and r.status == "complete"),
+        None,
+    )
+    sex = profile.sex if profile else None
+    if sex is None and prior:
+        sex = prior.details.get("sex")
+    row = QuestionnaireRecord(
+        id=uuid4().hex,
+        token=ctx.token,
+        kind=kind,
+        version=content()["version"],
+        policy_version=load_policy().version,
+        started_at=ctx.now,
+        day=ctx.now.astimezone(ZoneInfo(runtime.timezone)).date(),
+        status="active",
+        answers={},
+        details={
+            "runtime_version": runtime.version,
+            "modules": list(profile.modules or []) if profile else [],
+            "sex": sex,
+            "medication_active": bool(
+                (profile and profile.therapy_start and profile.therapy_start <= ctx.now.date())
+                or (prior and prior.answers.get("psychotropics") == 1)
+            ),
+            "suicide_monitoring": bool(profile and profile.enhanced_monitoring),
+            "weekly_pcl": kind == "pcl5" and bool((ctx.data.get("crisis") or {}).get("closed_at")),
+            "answer_metadata": {},
+            "task_id": task.id if task else None,
+        },
+    )
+    ctx.session.add(row)
+    await ctx.session.flush()
+    if task:
+        task.record_id = row.id
+        task.status = "active"
+    return row
+
+
+def save_answer(row, question, value, now):
+    """Anchor recall windows to administration, never to the day of a resumed answer."""
+    day = row.day
+    starts = {
+        "past_month": month_before(day),
+        "past_3_months": month_before(day, 3),
+        "past_week": day - timedelta(days=7),
+        "past_day": day - timedelta(days=1),
+        "последние 2 недели": day - timedelta(days=14),
+        "последний месяц": month_before(day),
+        "последняя неделя": day - timedelta(days=7),
+        "прошлая неделя": day - timedelta(days=7),
+        "последние 12 месяцев": month_before(day, 12),
+    }
+    row.answers = {**row.answers, question.id: value}
+    if question.id == "sex" and value in ("m", "f"):
+        row.details = {**row.details, "sex": value}
+    row.details = {
+        **row.details,
+        "answer_metadata": {
+            **row.details.get("answer_metadata", {}),
+            question.id: {
+                "answered_at": now.isoformat(),
+                "period": question.period,
+                "period_start": starts[question.period].isoformat() if question.period in starts else None,
+                "period_end": day.isoformat(),
+                "source": "patient",
+                "derived": False,
+            },
+        },
+    }
+
+
+def decision(row):
+    a = row.answers
+    sex = row.details.get("sex") or (a.get("sex") if a.get("sex") in ("m", "f") else None)
+    return evaluate_intake(
+        phq9=group_answers(a, "phq9"),
+        tsq=group_answers(a, "tsq"),
+        pcl5=group_answers(a, "pcl5"),
+        alcohol=group_answers(a, "rus_audit"),
+        cssrs=cssrs_answers(a),
+        sex=sex,
+        trauma_exposure=bool(a["trauma"]) if "trauma" in a else None,
+        traumatic_brain_injury=a.get("tbi") == 1,
+        psychotropics=a.get("psychotropics") == 1,
+    )
+
+
+def summary(row):
+    a = row.answers
+    totals = {}
+    for sid in ("phq9", "tsq", "pcl5"):
+        result = score(sid, group_answers(a, sid))
+        totals[sid] = {
+            "complete": result.complete,
+            "total": result.total if result.complete else None,
+            "answered": result.answered,
+        }
+    sex = row.details.get("sex") or (a.get("sex") if a.get("sex") in ("m", "f") else None)
+    for sid in ("rus_audit_s", "rus_audit"):
+        values = {k: a[k] for k in load_scale(sid).item_ids if k in a}
+        totals[sid] = asdict(score_alcohol(values, short=sid.endswith("_s"), sex=sex))
+    totals["cssrs_recent"] = asdict(score_cssrs_recent(cssrs_answers(a)))
+    return totals
+
+
+async def finish(ctx, row, status, *, extra=None):
+    row.status = status
+    row.completed_at = ctx.now if status == "complete" else None
+    row.details = {**row.details, **(extra or {})}
+    if row.details.get("task_id"):
+        task = await ctx.session.get(QuestionnaireTask, row.details["task_id"])
+        if task:
+            task.status = "done"
+
+
+async def alert(session, token, key, code, now, *, level=AlertLevel.MODERATE, params=None):
+    existing = await session.scalar(
+        select(QuestionnaireNotice).where(QuestionnaireNotice.token == token, QuestionnaireNotice.event_key == key)
+    )
+    if existing:
+        return existing
+    row = await repo.add_alert(session, token=token, level=level, code=code, reason=code, ts=now, params=params)
+    notice = QuestionnaireNotice(token=token, event_key=key, alert_id=row.id, created_at=now)
+    session.add(notice)
+    await session.flush()
+    return notice
+
+
+async def cancel_pending(session, token):
+    for row in await records(session, token):
+        if row.status in ("active", "paused"):
+            row.status = "cancelled"
+    tasks = await session.scalars(
+        select(QuestionnaireTask).where(
+            QuestionnaireTask.token == token, QuestionnaireTask.status.in_(("pending", "active"))
+        )
+    )
+    for task in tasks:
+        task.status = "cancelled"
+        task.delivery = []
+    notices = await session.scalars(
+        select(QuestionnaireNotice).where(QuestionnaireNotice.token == token, QuestionnaireNotice.status == "pending")
+    )
+    for notice in notices:
+        notice.status = "cancelled"
+
+
+async def dispatch_notices(token=None):
+    now = clock.now()
+    async with get_clinical_session() as session:
+        stmt = (
+            select(QuestionnaireNotice.id, QuestionnaireNotice.token)
+            .where(QuestionnaireNotice.status == "pending")
+            .order_by(QuestionnaireNotice.id)
+            .limit(100)
+        )
+        if token:
+            stmt = stmt.where(QuestionnaireNotice.token == token)
+        candidates = (await session.execute(stmt)).all()
+    for notice_id, patient_token in candidates:
+        async with get_clinical_session() as session:
+            # Same lock order as withdrawal: patient first, then notice.
+            await lock_patient(session, patient_token)
+            notice = await session.scalar(
+                select(QuestionnaireNotice).where(QuestionnaireNotice.id == notice_id).with_for_update()
+            )
+            if notice.status != "pending":
+                continue
+            if notice.attempted_at and now < notice.attempted_at + timedelta(
+                minutes=min(60, 2 ** min(notice.attempts, 6))
+            ):
+                continue
+            state = await repo.get_dialog_state(session, notice.token)
+            if state is None or state.data.get("consent") is not True:
+                notice.status = "cancelled"
+                continue
+            profile = await repo.get_profile(session, notice.token)
+            if not profile or not profile.doctor_id:
+                continue
+            item = await session.get(Alert, notice.alert_id)
+            notice.attempted_at, notice.attempts = now, notice.attempts + 1
+            try:
+                await asyncio.wait_for(
+                    integrations.notifier.send(profile.doctor_id, item.level, item.code, token=notice.token), timeout=10
+                )
+            except Exception as exc:
+                notice.error_code = type(exc).__name__[:80]
+            else:
+                notice.error_code = None
+                notice.accepted_at = now
+                notice.status = (
+                    "test_sink" if isinstance(integrations.notifier, MemoryNotifier) else "accepted_by_adapter"
+                )
+
+
+async def queue_task(session, token, kind, key, now):
+    existing = await session.scalar(
+        select(QuestionnaireTask).where(QuestionnaireTask.token == token, QuestionnaireTask.task_key == key)
+    )
+    if existing:
+        return existing
+    # Only one outstanding administration per form, however many ticks were missed.
+    existing = await session.scalar(
+        select(QuestionnaireTask).where(
+            QuestionnaireTask.token == token,
+            QuestionnaireTask.kind == kind,
+            QuestionnaireTask.status.in_(("pending", "active")),
+        )
+    )
+    if existing:
+        return existing
+    row = QuestionnaireTask(token=token, kind=kind, task_key=key, due_at=now)
+    session.add(row)
+    await session.flush()
+    return row
+
+
+async def lock_patient(session, token):
+    # PostgreSQL serializes callbacks from different workers. SQLite tests additionally
+    # use the engine's in-process lock. Consent and answers share this transaction.
+    await session.scalar(select(DialogState).where(DialogState.token == token).with_for_update())