Arxip222andClaude Opus 5.5
290b0d6151
- Sign-in: centred ЕМИАС card without the sidebar; anonymous entry asks «Депрессия / ПТСР».
- Stage 1 is the same for everyone and follows the fixed order of the MVP questionnaire:
PHQ-2 → trauma filter/TSQ → RUS-AUDIT-S → C-SSRS → context; stage 2 only where stage 1
is positive. Starts without a profile.
- After stage 1 the patient sees a preliminary result (screening, not a diagnosis), stored
with the record; it switches on the diary modules (depression / PTSD).
- Stage 2 diary: morning (sleep, sleep quality, alcohol) and evening (mood, medication,
3 PHQ-9 items in rotation, short PCL-5); C-SSRS since last contact only under risk;
flags on weekly means against the personal norm; scheduled full scales with change
(remission, response, worsening).
- Questionnaire log: GET /me/{token}/results and the «Мои результаты» screen.
- Questionnaires centred without the sidebar and without scale names; «Продолжить позже»
goes home; confirmations are toasts; free-text comments are optional.
- Demo policy config/questionnaire_demo.yaml (INTAKE_DEMO only, not a clinical approval).
ADR-012.
Refs #77
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
348 lines
15 KiB
Python
348 lines
15 KiB
Python
"""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
|
|
|
|
import os
|
|
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 demo_mode() -> bool:
|
|
"""Demo-only switch (INTAKE_DEMO=1 with the virtual clock): questionnaires on synthetic data.
|
|
|
|
It selects `config/questionnaire_demo.yaml` (marked as NOT a clinical approval) instead of the real policy; the
|
|
real policy and the release gate in config/intake_runtime.yaml are never touched.
|
|
"""
|
|
return os.environ.get("INTAKE_DEMO", "").lower() in ("1", "true", "yes") and get_settings().demo_clock
|
|
|
|
|
|
def load_policy(path: Path | None = None) -> IntakePolicy:
|
|
path = path or get_settings().config_dir / ("questionnaire_demo.yaml" if demo_mode() else "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), ())
|