From 711c646876b9070c6aa65547bf6621d5e9e0b48b Mon Sep 17 00:00:00 2001 From: Lacky1234union Date: Sat, 3 Oct 2026 14:58:18 +0300 Subject: [PATCH] feat(core): rule engine for alerts/agent actions and virtual demo clock - app/core/clock.py: now()/today(), virtual offset (advance/set_now/reset), pluggable offset persistence hook (ADR-005) - config/rules.yaml + app/core/rules.py: YAML declares rules (params, actions, levels, reason_ru), Python evaluators implement predicates by id; dedup of open alerts, crisis never suppressed; ADR-006 colours (ADR-007) - tests: per-rule unit tests, dedup, clock, 13-day demo persona trajectory Refs #5 Co-Authored-By: Claude Opus 5.5 --- app/core/clock.py | 122 +++++++++++ app/core/rules.py | 481 ++++++++++++++++++++++++++++++++++++++++++++ config/rules.yaml | 178 ++++++++++++++++ tests/test_clock.py | 70 +++++++ tests/test_rules.py | 427 +++++++++++++++++++++++++++++++++++++++ 5 files changed, 1278 insertions(+) create mode 100644 app/core/clock.py create mode 100644 app/core/rules.py create mode 100644 config/rules.yaml create mode 100644 tests/test_clock.py create mode 100644 tests/test_rules.py diff --git a/app/core/clock.py b/app/core/clock.py new file mode 100644 index 0000000..587b7d8 --- /dev/null +++ b/app/core/clock.py @@ -0,0 +1,122 @@ +"""The only source of time in the system (ADR-005, guardrail #4). + +Real mode (`DEMO_CLOCK=false`): `now()` is the wall clock in `settings.tz`. +Virtual mode (`DEMO_CLOCK=true`): `now()` is the wall clock shifted by an offset. The demo console moves +time forward with `advance(days=1)` ("next day"), so 13 days of therapy fit into a 3-minute demo while the +scheduler and the rule windows ("3 days in a row", "7 days") keep working on the shifted time. + +The clock keeps an *offset*, not a frozen instant: virtual time keeps ticking between button presses, so +scheduled jobs and timestamps remain ordered. The offset can be persisted through a pluggable hook +(`register_persistence`) so the DB layer can store it and survive restarts; by default it lives in memory. + +Business logic must never call `datetime.now()` directly — it receives time from here (or as a parameter). +""" + +from __future__ import annotations + +import threading +from collections.abc import Callable +from datetime import date, datetime, timedelta, timezone, tzinfo +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError + +from app.config import get_settings + +LoadOffset = Callable[[], timedelta | None] +SaveOffset = Callable[[timedelta], None] + +_lock = threading.RLock() +_offset = timedelta(0) +_save: SaveOffset | None = None + +# Moscow has had a fixed UTC+3 without DST since 2014; used only if the OS has no tz database +# (slim Docker images may ship without tzdata). +_FALLBACK_OFFSETS = {"Europe/Moscow": timedelta(hours=3)} + + +def _tz() -> tzinfo: + name = get_settings().tz + try: + return ZoneInfo(name) + except ZoneInfoNotFoundError: + return timezone(_FALLBACK_OFFSETS.get(name, timedelta(0)), name) + + +def _virtual() -> bool: + return get_settings().demo_clock + + +def _real_now() -> datetime: + # The single sanctioned wall-clock read in the codebase. + return datetime.now(_tz()) + + +def _require_virtual() -> None: + if not _virtual(): + raise RuntimeError("Virtual clock is disabled (DEMO_CLOCK=false); time cannot be moved") + + +def _store(offset: timedelta) -> None: + global _offset + _offset = offset + if _save is not None: + _save(offset) + + +def now() -> datetime: + """Current time, tz-aware in `settings.tz`; shifted by the demo offset in virtual mode.""" + real = _real_now() + if not _virtual(): + return real + with _lock: + return real + _offset + + +def today() -> date: + """Current local date according to `now()`.""" + return now().date() + + +def offset() -> timedelta: + """Current virtual offset (always zero in real mode).""" + if not _virtual(): + return timedelta(0) + with _lock: + return _offset + + +def advance(days: float = 0, hours: float = 0, minutes: float = 0) -> datetime: + """Move virtual time forward (or backward with negative values). Returns the new `now()`.""" + _require_virtual() + with _lock: + _store(_offset + timedelta(days=days, hours=hours, minutes=minutes)) + return now() + + +def set_now(dt: datetime) -> datetime: + """Jump virtual time to `dt` (naive values are interpreted in `settings.tz`). Returns the new `now()`.""" + _require_virtual() + if dt.tzinfo is None: + dt = dt.replace(tzinfo=_tz()) + with _lock: + _store(dt - _real_now()) + return now() + + +def reset() -> None: + """Return virtual time to the wall clock (offset = 0).""" + with _lock: + _store(timedelta(0)) + + +def register_persistence(load: LoadOffset | None, save: SaveOffset | None) -> None: + """Plug in offset storage (e.g. a DB row). `load` is called once immediately; `save` on every change. + + Pass `(None, None)` to go back to in-memory only. + """ + global _offset, _save + with _lock: + _save = save + if load is not None: + stored = load() + if stored is not None: + _offset = stored diff --git a/app/core/rules.py b/app/core/rules.py new file mode 100644 index 0000000..a619192 --- /dev/null +++ b/app/core/rules.py @@ -0,0 +1,481 @@ +"""Deterministic alert & agent-action engine (ADR-007, guardrail #3). + +The LLM never decides whether to alert. Structured data (daily check-ins, doses, scale totals, C-SSRS) is +packed into a `PatientSnapshot`; `evaluate()` runs the rules and returns `Action`s that the orchestrator +executes (create an alert, start the crisis script, schedule a scale, send medication info...). The LLM only +phrases the reply around an action that has already been chosen. + +How a rule is defined — the YAML/code split: + * `config/rules.yaml` declares each rule: `id`, description, thresholds/windows (`params`), the actions it + produces (type, level, extra params) and `reason_ru` for the doctor. Clinicians edit only this file. + * This module implements the predicate of each rule as a pure function registered under the same id with + `@evaluator("")`. It returns `None` when the rule does not fire, or a dict of computed values + (e.g. `{"total": 22}`) that is merged into action params and used to format `reason_ru`. + * Loading validates both directions: a YAML id without an evaluator is an error (no silently dead rules). + +Determinism: evaluators read time only from `snapshot.as_of` (the virtual "now" from `app.core.clock`), +never from the wall clock, so the same snapshot always yields the same actions. + +Dedup: when a rule would emit an `alert` whose code is already open (not acked by the doctor) the whole rule +stays silent. `crisis` actions are never suppressed — the orchestrator is responsible for not restarting an +already running crisis script. + +Colours (ADR-006): ACUTE → red; HIGH/MODERATE → yellow; INFO only → grey; no open alerts → green. +""" + +from __future__ import annotations + +from collections.abc import Callable, Iterable +from datetime import date, datetime, timedelta +from functools import lru_cache +from pathlib import Path +from typing import Any, Literal + +import yaml +from pydantic import BaseModel, Field, field_validator, model_validator + +from app.config import get_settings +from app.core.types import AlertLevel + +# ---------------------------------------------------------------------------------------------- snapshot + +Module = Literal["veteran", "depression", "family"] +SleepBucket = Literal["<4", "4-6", "6-8", ">8"] +AlcoholBucket = Literal["none", "1-2", "more"] +DoseStatus = Literal["taken", "missed", "skipped", "pending"] +CssrsRisk = Literal["none", "low", "moderate", "high"] +ActionType = Literal[ + "alert", + "crisis", + "schedule_scale", + "reach_out", + "med_info", + "suggest_visit", + "show_safety_plan", + "enhanced_monitoring", +] +StatusColor = Literal["red", "yellow", "green", "grey"] + +_CSSRS_RANK: dict[str, int] = {"none": 0, "low": 1, "moderate": 2, "high": 3} +_LEVELS: dict[str, AlertLevel] = { + "info": AlertLevel.INFO, + "moderate": AlertLevel.MODERATE, + "high": AlertLevel.HIGH, + "acute": AlertLevel.ACUTE, +} + + +class DailyCheckin(BaseModel): + """One day of the daily check-in (ADR-002). `None` = not asked/not answered that day.""" + + day: date + mood: int | None = Field(default=None, ge=0, le=10) + med_taken: bool | None = None + sleep_hours_bucket: SleepBucket | None = None + alcohol: AlcoholBucket | None = None + side_effects: list[str] = Field(default_factory=list) # normalised codes from `extract`, e.g. "nausea" + quit_intent: bool = False # "I'll stop taking the pills" detected in free text + nightmares: bool | None = None + + +class ScalePoint(BaseModel): + """A completed scale: only the total (+ PHQ-9 item 9) is needed by the rules.""" + + scale: str # phq9 | pcl5 | gad7 | auditc | ... + ts: datetime + total: int = Field(ge=0) + item9: int | None = Field(default=None, ge=0, le=3) + scheduled: bool = True # False = unscheduled (triggered by a rule) + + +class DoseRecord(BaseModel): + planned_at: datetime + status: DoseStatus + + +class PatientSnapshot(BaseModel): + """Everything the rules may look at, assembled by the caller from the clinical DB (pseudonymised).""" + + token: str + as_of: datetime # virtual "now" (app.core.clock.now()); all windows are counted back from here + age: int = Field(ge=0) + sex: Literal["m", "f"] + modules: set[Module] = Field(default_factory=set) + therapy_day: int = 0 # day 0 = visit where therapy was prescribed + checkins: list[DailyCheckin] = Field(default_factory=list) + scale_results: list[ScalePoint] = Field(default_factory=list) + doses: list[DoseRecord] = Field(default_factory=list) + cssrs_risk: CssrsRisk | None = None + weapon_access: bool | None = None + last_patient_message_at: datetime | None = None + open_alert_codes: set[str] = Field(default_factory=set) # alerts not yet acked by the doctor + + @field_validator("as_of") + @classmethod + def _aware(cls, v: datetime) -> datetime: + if v.tzinfo is None: + raise ValueError("as_of must be timezone-aware (take it from app.core.clock.now())") + return v + + +class Action(BaseModel): + """One output of the rule engine, executed by the orchestrator.""" + + type: ActionType + level: AlertLevel | None = None + code: str # stable reason code = rule id (dedup key, stored with the alert) + params: dict[str, Any] = Field(default_factory=dict) + reason_ru: str = "" # short text for the doctor dashboard + + @property + def creates_alert(self) -> bool: + """`alert` and `crisis` actions are persisted as an alert record (crisis = acute alert).""" + return self.type in ("alert", "crisis") + + +# ---------------------------------------------------------------------------------------------- config + + +class ActionSpec(BaseModel): + type: ActionType + level: Literal["info", "moderate", "high", "acute"] | None = None + params: dict[str, Any] = Field(default_factory=dict) + + @model_validator(mode="after") + def _alert_needs_level(self) -> ActionSpec: + if self.type in ("alert", "crisis") and self.level is None: + raise ValueError(f"action '{self.type}' requires a level") + return self + + +class RuleSpec(BaseModel): + id: str + description: str = "" + enabled: bool = True + params: dict[str, Any] = Field(default_factory=dict) + actions: list[ActionSpec] = Field(min_length=1) + reason_ru: str = "" + + +class RulesConfig(BaseModel): + rules: list[RuleSpec] + + @model_validator(mode="after") + def _check_ids(self) -> RulesConfig: + ids = [r.id for r in self.rules] + dupes = {i for i in ids if ids.count(i) > 1} + if dupes: + raise ValueError(f"duplicate rule ids: {sorted(dupes)}") + unknown = set(ids) - set(_REGISTRY) + if unknown: + raise ValueError(f"rules without an evaluator in app/core/rules.py: {sorted(unknown)}") + return self + + +def load_rules(path: Path | None = None) -> RulesConfig: + """Parse and validate a rules YAML (default: `/rules.yaml`, cached).""" + if path is None: + return _default_rules() + return RulesConfig.model_validate(yaml.safe_load(path.read_text(encoding="utf-8"))) + + +@lru_cache +def _default_rules() -> RulesConfig: + return load_rules(get_settings().config_dir / "rules.yaml") + + +# ---------------------------------------------------------------------------------------------- registry + +Context = dict[str, Any] +Evaluator = Callable[[PatientSnapshot, dict[str, Any]], Context | None] +_REGISTRY: dict[str, Evaluator] = {} + + +def evaluator(rule_id: str) -> Callable[[Evaluator], Evaluator]: + """Register the predicate implementing YAML rule `rule_id`.""" + + def deco(fn: Evaluator) -> Evaluator: + if rule_id in _REGISTRY: + raise ValueError(f"evaluator for '{rule_id}' already registered") + _REGISTRY[rule_id] = fn + return fn + + return deco + + +def registered_rule_ids() -> set[str]: + return set(_REGISTRY) + + +# ---------------------------------------------------------------------------------------------- helpers + + +def _today(s: PatientSnapshot) -> date: + return s.as_of.date() + + +def _checkins_by_day(s: PatientSnapshot) -> dict[date, DailyCheckin]: + return {c.day: c for c in s.checkins if c.day <= _today(s)} + + +def _anchor_day(s: PatientSnapshot) -> date: + """Last day a "N days in a row" streak may end on. + + Today if today's check-in exists, otherwise yesterday — so a morning evaluation (before today's check-in) + still sees yesterday's streak, but an old streak does not keep firing forever. + """ + today = _today(s) + return today if today in _checkins_by_day(s) else today - timedelta(days=1) + + +def _streak(s: PatientSnapshot, days: int, pred: Callable[[DailyCheckin], bool]) -> list[DailyCheckin] | None: + """Check-ins of `days` consecutive calendar days ending at the anchor day, if all satisfy `pred`. + + A missing day breaks the streak: we never assume a symptom on a day the patient did not report. + """ + by_day = _checkins_by_day(s) + end = _anchor_day(s) + run: list[DailyCheckin] = [] + for i in range(days): + c = by_day.get(end - timedelta(days=i)) + if c is None or not pred(c): + return None + run.append(c) + return run + + +def _history(s: PatientSnapshot, scale: str) -> list[ScalePoint]: + """Results of `scale` up to `as_of`, oldest first (results "from the future" are ignored).""" + return sorted((p for p in s.scale_results if p.scale == scale and p.ts <= s.as_of), key=lambda p: p.ts) + + +def _recent_item9(s: PatientSnapshot, threshold: int, recent_hours: float) -> int | None: + """Max PHQ-9 item 9 among PHQ-9s completed within `recent_hours`, if it reaches `threshold`. + + Limited to recent results so one old positive answer does not restart the crisis protocol on every + evaluation; after that the open acute alert keeps the patient red until the doctor acts. + """ + since = s.as_of - timedelta(hours=recent_hours) + values = [p.item9 for p in _history(s, "phq9") if p.ts > since and p.item9 is not None] + top = max(values, default=None) + return top if top is not None and top >= threshold else None + + +# ---------------------------------------------------------------------------------------------- evaluators +# Each function implements the predicate of the YAML rule with the same id; thresholds come from `p`. + + +@evaluator("phq9_item9") +def _phq9_item9(s: PatientSnapshot, p: dict[str, Any]) -> Context | None: + # Item 9 is collected only by buttons (never LLM-interpreted), so the value is the patient's own answer. + item9 = _recent_item9(s, p["threshold"], p["recent_hours"]) + return None if item9 is None else {"item9": item9} + + +@evaluator("cssrs_high") +def _cssrs_high(s: PatientSnapshot, p: dict[str, Any]) -> Context | None: + return {} if s.cssrs_risk == "high" else None + + +@evaluator("weapon_access_risk") +def _weapon_access_risk(s: PatientSnapshot, p: dict[str, Any]) -> Context | None: + # Access to means alone is not a crisis; combined with any suicide-risk signal it is. + if not s.weapon_access: + return None + item9_signal = _recent_item9(s, p["item9_threshold"], p["recent_hours"]) is not None + cssrs_signal = _CSSRS_RANK.get(s.cssrs_risk or "none", 0) >= _CSSRS_RANK[p["cssrs_min"]] + return {} if item9_signal or cssrs_signal else None + + +@evaluator("cssrs_moderate") +def _cssrs_moderate(s: PatientSnapshot, p: dict[str, Any]) -> Context | None: + return {} if s.cssrs_risk == "moderate" else None + + +def _latest_at_least(scale: str) -> Evaluator: + """Latest total of `scale` >= params.threshold.""" + + def fn(s: PatientSnapshot, p: dict[str, Any]) -> Context | None: + hist = _history(s, scale) + if hist and hist[-1].total >= p["threshold"]: + return {"total": hist[-1].total} + return None + + return fn + + +def _rise_between_last_two(scale: str) -> Evaluator: + """Latest total minus the previous total of `scale` >= params.min_rise (worsening).""" + + def fn(s: PatientSnapshot, p: dict[str, Any]) -> Context | None: + hist = _history(s, scale) + if len(hist) < 2: + return None + prev, last = hist[-2].total, hist[-1].total + if last - prev >= p["min_rise"]: + return {"previous": prev, "total": last, "rise": last - prev} + return None + + return fn + + +evaluator("phq9_severe")(_latest_at_least("phq9")) +evaluator("pcl5_severe")(_latest_at_least("pcl5")) +evaluator("phq9_rise")(_rise_between_last_two("phq9")) +evaluator("pcl5_rise")(_rise_between_last_two("pcl5")) + + +@evaluator("auditc_positive") +def _auditc_positive(s: PatientSnapshot, p: dict[str, Any]) -> Context | None: + # Sex-specific cut-off: >= 4 for men, >= 3 for women (lower tolerance / higher risk at equal intake). + hist = _history(s, "auditc") + threshold = p["threshold_m"] if s.sex == "m" else p["threshold_f"] + if hist and hist[-1].total >= threshold: + return {"total": hist[-1].total, "threshold": threshold} + return None + + +@evaluator("quit_intent") +def _quit_intent(s: PatientSnapshot, p: dict[str, Any]) -> Context | None: + # Fires on the day the intent is voiced, and on a later skipped dose within the window + # (the intent turned into action). + anchor = _anchor_day(s) + today = _checkins_by_day(s).get(anchor) + if today is None: + return None + if today.quit_intent: + return {} + since = anchor - timedelta(days=p["window_days"]) + intent_before = any(c.quit_intent for c in s.checkins if since <= c.day < anchor) + return {} if today.med_taken is False and intent_before else None + + +@evaluator("missed_doses_7d") +def _missed_doses(s: PatientSnapshot, p: dict[str, Any]) -> Context | None: + # Counts distinct calendar days (not doses) within the window with a missed/skipped dose, + # from dose records and from the check-in answer "did not take" — whichever source has the data. + window = p["window_days"] + first = _today(s) - timedelta(days=window - 1) + tz = s.as_of.tzinfo + days = { + d.planned_at.astimezone(tz).date() + for d in s.doses + if d.status in ("missed", "skipped") and d.planned_at <= s.as_of + } + days |= {c.day for c in s.checkins if c.med_taken is False} + count = sum(1 for d in days if first <= d <= _today(s)) + return {"count": count} if count >= p["threshold"] else None + + +@evaluator("side_effect_3d") +def _side_effect_3d(s: PatientSnapshot, p: dict[str, Any]) -> Context | None: + run = _streak(s, p["consecutive_days"], lambda c: bool(c.side_effects)) + if run is None: + return None + effects = sorted({e for c in run for e in c.side_effects}) + return {"effects": ", ".join(effects)} + + +@evaluator("sleep_lt4_3n") +def _sleep_lt4(s: PatientSnapshot, p: dict[str, Any]) -> Context | None: + return None if _streak(s, p["consecutive_days"], lambda c: c.sleep_hours_bucket == "<4") is None else {} + + +@evaluator("alcohol_3d") +def _alcohol_3d(s: PatientSnapshot, p: dict[str, Any]) -> Context | None: + drank = _streak(s, p["consecutive_days"], lambda c: c.alcohol is not None and c.alcohol != "none") + return None if drank is None else {} + + +@evaluator("mood_low_3d") +def _mood_low_3d(s: PatientSnapshot, p: dict[str, Any]) -> Context | None: + low = _streak(s, p["consecutive_days"], lambda c: c.mood is not None and c.mood <= p["threshold"]) + if low is None: + return None + # Do not ask for an unscheduled PHQ-9 again if one was already taken recently. + since = s.as_of - timedelta(days=p["cooldown_days"]) + if any(not r.scheduled and r.ts > since for r in _history(s, "phq9")): + return None + return {} + + +@evaluator("no_contact_3d") +def _no_contact(s: PatientSnapshot, p: dict[str, Any]) -> Context | None: + if s.last_patient_message_at is None: + return None + silence = s.as_of - s.last_patient_message_at + return {"silent_days": silence.days} if silence > timedelta(days=p["days"]) else None + + +@evaluator("young_adult_start") +def _young_adult_start(s: PatientSnapshot, p: dict[str, Any]) -> Context | None: + if s.age < p["age_below"] and 0 <= s.therapy_day <= p["max_therapy_day"]: + return {} + return None + + +# ---------------------------------------------------------------------------------------------- engine + + +class _KeepMissing(dict[str, Any]): + def __missing__(self, key: str) -> str: + return "{" + key + "}" + + +def _build_actions(rule: RuleSpec, ctx: Context) -> list[Action]: + reason = rule.reason_ru.format_map(_KeepMissing({**rule.params, **ctx})) + return [ + Action( + type=spec.type, + level=_LEVELS[spec.level] if spec.level else None, + code=rule.id, + params={**ctx, **spec.params}, + reason_ru=reason, + ) + for spec in rule.actions + ] + + +def evaluate(snapshot: PatientSnapshot, config: RulesConfig | None = None) -> list[Action]: + """Run all enabled rules (YAML order) on the snapshot. Pure and deterministic.""" + cfg = config or load_rules() + out: list[Action] = [] + for rule in cfg.rules: + if not rule.enabled: + continue + ctx = _REGISTRY[rule.id](snapshot, rule.params) + if ctx is None: + continue + actions = _build_actions(rule, ctx) + is_crisis = any(a.type == "crisis" for a in actions) + already_open = any(a.type == "alert" and a.code in snapshot.open_alert_codes for a in actions) + if already_open and not is_crisis: + continue + out.extend(actions) + return out + + +# ---------------------------------------------------------------------------------------------- colours + + +class PatientStatus(BaseModel): + color: StatusColor + top_level: AlertLevel | None + + +def color(levels: Iterable[AlertLevel | None]) -> StatusColor: + """Dashboard colour for a set of open alert levels (ADR-006).""" + present = {lvl for lvl in levels if lvl is not None} + if AlertLevel.ACUTE in present: + return "red" + if present & {AlertLevel.HIGH, AlertLevel.MODERATE}: + return "yellow" + if AlertLevel.INFO in present: + return "grey" + return "green" + + +def patient_status(open_alert_levels: Iterable[AlertLevel | None]) -> PatientStatus: + """Colour + highest open level for a patient row on the dashboard.""" + levels = [lvl for lvl in open_alert_levels if lvl is not None] + return PatientStatus(color=color(levels), top_level=max(levels, default=None)) diff --git a/config/rules.yaml b/config/rules.yaml new file mode 100644 index 0000000..1f19944 --- /dev/null +++ b/config/rules.yaml @@ -0,0 +1,178 @@ +# Alert & agent-action rules (ADR-007, ADR-008). +# +# Split of responsibilities: +# * This file DECLARES the rules: id, clinical description, thresholds/windows (`params`), which actions the +# rule produces (type, level, extra params) and the short Russian reason shown on the doctor dashboard. +# A psychiatrist can change thresholds, levels, actions and texts here without touching code. +# * `app/core/rules.py` IMPLEMENTS the predicate of each rule as a Python function registered under the +# same `id` (`@evaluator("")`). The function reads only the patient snapshot and `params` from here. +# Adding a new rule = new YAML entry + new evaluator; an id without an evaluator fails at load time. +# +# Fields: +# id — stable rule id; also the default alert/reason code (dedup key, stored with the alert). +# description — what and why (for the reviewer/jury); not shown to patients. +# enabled — false switches the rule off without deleting it. +# params — thresholds/windows passed to the evaluator; also available in `reason_ru` as {name}. +# actions — produced in this order when the predicate fires. `level` is info|moderate|high|acute. +# Action `params` are merged with values computed by the evaluator. +# reason_ru — short text for the doctor (str.format with params + evaluator values). +# +# Dedup: if a rule produces an `alert` whose code is already open (not acked by the doctor), the whole rule +# is silent — no repeated alert, no repeated med_info/reach_out. `crisis` is never suppressed. +# +# Starting thresholds are from SPEC §6, CLINICAL_MODULES.md and the demo persona; final values are approved +# by the team psychiatrist. + +rules: + # ---------------------------------------------------------------- suicide risk → crisis (red, ADR-006) + - id: phq9_item9 + description: > + PHQ-9 item 9 (thoughts of death/self-harm) >= 1 in a recent PHQ-9. Always a crisis: fixed safety + script, safety plan on screen, C-SSRS screening, immediate escalation to the doctor on duty. + params: {threshold: 1, recent_hours: 24} + actions: + - {type: crisis, level: acute} + - {type: show_safety_plan} + - {type: schedule_scale, params: {scale: cssrs_screen, scheduled: false}} + reason_ru: "PHQ-9 п.9 = {item9} (мысли о смерти/самоповреждении)" + + - id: cssrs_high + description: > + C-SSRS: suicidal ideation with intent/plan or suicidal behaviour (risk "high") → urgent escalation, + safety plan and crisis contacts on screen. + params: {} + actions: + - {type: crisis, level: acute} + - {type: show_safety_plan} + reason_ru: "C-SSRS: высокий риск (намерение/план или поведение)" + + - id: weapon_access_risk + description: > + Access to weapons together with any suicide-risk signal (recent PHQ-9 item 9 >= 1 or C-SSRS >= moderate). + Means restriction is one of the most evidence-based suicide-prevention measures → crisis safety block. + params: {item9_threshold: 1, cssrs_min: moderate, recent_hours: 24} + actions: + - {type: crisis, level: acute} + - {type: show_safety_plan, params: {block: means_restriction}} + reason_ru: "Доступ к оружию при сигнале суицидального риска" + + - id: cssrs_moderate + description: > + C-SSRS moderate (ideation without plan/intent) = SPEC "Safety POSSIBLE" → doctor within 24 h. + params: {} + actions: + - {type: alert, level: high} + reason_ru: "C-SSRS: умеренный риск, врачу в течение 24 ч" + + # ---------------------------------------------------------------- scale thresholds (yellow) + - id: phq9_severe + description: Latest PHQ-9 total in the severe range (>= 20) → doctor within 24 h. + params: {threshold: 20} + actions: + - {type: alert, level: high} + reason_ru: "PHQ-9 = {total} (тяжёлая депрессия, порог {threshold})" + + - id: pcl5_severe + description: Latest PCL-5 total >= 50 → doctor within 24 h. + params: {threshold: 50} + actions: + - {type: alert, level: high} + reason_ru: "PCL-5 = {total} (порог {threshold})" + + - id: phq9_rise + description: PHQ-9 rose by >= 5 points between the two latest measurements → doctor within 24 h. + params: {min_rise: 5} + actions: + - {type: alert, level: high} + reason_ru: "Рост PHQ-9 {previous} → {total} (+{rise})" + + - id: pcl5_rise + description: PCL-5 rose by >= 10 points between the two latest measurements (veteran module) → doctor. + params: {min_rise: 10} + actions: + - {type: alert, level: high} + reason_ru: "Рост PCL-5 {previous} → {total} (+{rise})" + + - id: auditc_positive + description: > + AUDIT-C positive screen (>= 4 men / >= 3 women). Alcohol is a suicide risk factor and interacts with + antidepressants → flag for the visit, daily alcohol question. + params: {threshold_m: 4, threshold_f: 3} + actions: + - {type: alert, level: moderate} + reason_ru: "AUDIT-C = {total} (положительный скрининг, порог {threshold})" + + # ---------------------------------------------------------------- medication adherence + - id: quit_intent + description: > + Patient says they want to stop the medication (extracted from free text), or skips the dose after having + expressed that intent within the window. Delayed SSRI effect (2–4 weeks) is the usual reason → explain, + advise not to stop without the doctor, notify the doctor. + params: {window_days: 7} + actions: + - {type: alert, level: high} + - {type: med_info, params: {topic: delayed_effect}} + reason_ru: "Намерение прекратить приём препарата" + + - id: missed_doses_7d + description: > + >= 3 days with a missed/skipped dose in the last 7 days → check-in about the reasons, flag for the visit. + params: {threshold: 3, window_days: 7} + actions: + - {type: alert, level: moderate} + - {type: reach_out, params: {topic: missed_doses}} + reason_ru: "Пропуски приёма: {count} из {window_days} дней" + + - id: side_effect_3d + description: > + Side-effect complaints on 3 consecutive days (e.g. SSRI nausea, persona day 4) → flag for the doctor. + params: {consecutive_days: 3} + actions: + - {type: alert, level: moderate} + reason_ru: "Побочные эффекты {consecutive_days} дня подряд: {effects}" + + # ---------------------------------------------------------------- daily check-in patterns + - id: sleep_lt4_3n + description: Sleep < 4 h for 3 consecutive nights (F_Y6) → flag for the doctor. + params: {consecutive_days: 3} + actions: + - {type: alert, level: moderate} + reason_ru: "Сон < 4 ч {consecutive_days} ночи подряд" + + - id: alcohol_3d + description: > + Alcohol on 3 consecutive days (F_Y5) → information about alcohol/antidepressant interaction, + flag for the doctor, suggest an earlier visit. + params: {consecutive_days: 3} + actions: + - {type: alert, level: moderate} + - {type: med_info, params: {topic: alcohol_interaction}} + - {type: suggest_visit, params: {reason: alcohol}} + reason_ru: "Алкоголь {consecutive_days} дня подряд" + + - id: mood_low_3d + description: > + Mood <= 3 (0–10) on 3 consecutive days → unscheduled PHQ-9 (ADR-002). Not an alert by itself; skipped if + an unscheduled PHQ-9 was already taken within the cooldown. + params: {threshold: 3, consecutive_days: 3, cooldown_days: 7} + actions: + - {type: schedule_scale, params: {scale: phq9, scheduled: false}} + reason_ru: "Настроение ≤ {threshold} {consecutive_days} дня подряд → внеплановый PHQ-9" + + - id: no_contact_3d + description: > + No message from the patient for > 3 days. Silence itself is a risk signal → grey flag, caring contact. + params: {days: 3} + actions: + - {type: alert, level: info} + - {type: reach_out, params: {topic: silence}} + reason_ru: "Нет ответа пациента {silent_days} дн." + + - id: young_adult_start + description: > + Age < 25 in the first 4 weeks of antidepressant therapy: possible increase of suicidal thoughts (FDA + boxed warning) → enhanced monitoring (more frequent check-ins and C-SSRS). + params: {age_below: 25, max_therapy_day: 28} + actions: + - {type: enhanced_monitoring, level: info} + reason_ru: "Возраст < 25, первые 4 недели терапии: усиленный мониторинг" diff --git a/tests/test_clock.py b/tests/test_clock.py new file mode 100644 index 0000000..e0b47be --- /dev/null +++ b/tests/test_clock.py @@ -0,0 +1,70 @@ +from collections.abc import Iterator +from datetime import datetime, timedelta + +import pytest + +from app.config import Settings +from app.core import clock + + +@pytest.fixture(autouse=True) +def _fresh_clock(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]: + monkeypatch.setattr(clock, "get_settings", lambda: Settings(_env_file=None, demo_clock=True)) + clock.register_persistence(None, None) + clock.reset() + yield + clock.register_persistence(None, None) + clock.reset() + + +def _close(a: datetime, b: datetime) -> bool: + return abs(a - b) < timedelta(seconds=5) + + +def test_now_is_tz_aware_in_settings_tz() -> None: + now = clock.now() + assert now.tzinfo is not None + assert now.utcoffset() == timedelta(hours=3) # Europe/Moscow + + +def test_advance_shifts_time() -> None: + before = clock.now() + after = clock.advance(days=1, hours=2, minutes=30) + assert _close(after, before + timedelta(days=1, hours=2, minutes=30)) + assert clock.offset() == timedelta(days=1, hours=2, minutes=30) + assert clock.today() == clock.now().date() + + +def test_set_now_and_reset() -> None: + target = datetime(2026, 10, 20, 9, 0) + assert _close(clock.set_now(target), target.replace(tzinfo=clock.now().tzinfo)) + clock.reset() + assert clock.offset() == timedelta(0) + assert _close(clock.now(), clock._real_now()) + + +def test_persistence_hook_loads_and_saves() -> None: + saved: list[timedelta] = [] + clock.register_persistence(lambda: timedelta(days=5), saved.append) + assert clock.offset() == timedelta(days=5) + clock.advance(days=1) + assert saved[-1] == timedelta(days=6) + clock.reset() + assert saved[-1] == timedelta(0) + + +def test_persistence_load_none_keeps_offset() -> None: + clock.advance(hours=3) + clock.register_persistence(lambda: None, None) + assert clock.offset() == timedelta(hours=3) + + +def test_real_mode_ignores_offset_and_forbids_moving(monkeypatch: pytest.MonkeyPatch) -> None: + clock.advance(days=3) + monkeypatch.setattr(clock, "get_settings", lambda: Settings(_env_file=None, demo_clock=False)) + assert clock.offset() == timedelta(0) + assert _close(clock.now(), clock._real_now()) + with pytest.raises(RuntimeError): + clock.advance(days=1) + with pytest.raises(RuntimeError): + clock.set_now(datetime(2026, 1, 1)) diff --git a/tests/test_rules.py b/tests/test_rules.py new file mode 100644 index 0000000..f18f9b5 --- /dev/null +++ b/tests/test_rules.py @@ -0,0 +1,427 @@ +from datetime import date, datetime, time, timedelta, timezone +from typing import Any + +import pytest +from pydantic import ValidationError + +from app.core.rules import ( + Action, + DailyCheckin, + DoseRecord, + PatientSnapshot, + RulesConfig, + ScalePoint, + color, + evaluate, + load_rules, + patient_status, + registered_rule_ids, +) +from app.core.types import AlertLevel + +MSK = timezone(timedelta(hours=3), "MSK") +DAY0 = date(2026, 9, 21) # visit to the psychiatrist, sertraline prescribed + + +def at(day: int, hour: int = 21) -> datetime: + return datetime.combine(DAY0 + timedelta(days=day), time(hour), MSK) + + +def snap(day: int, **kw: Any) -> PatientSnapshot: + base: dict[str, Any] = { + "token": "t" * 32, + "as_of": at(day), + "age": 28, + "sex": "m", + "modules": {"depression"}, + "therapy_day": day, + "last_patient_message_at": at(day, 20), + } + base.update(kw) + return PatientSnapshot(**base) + + +def ci(day: int, **kw: Any) -> DailyCheckin: + return DailyCheckin(day=DAY0 + timedelta(days=day), **kw) + + +def codes(actions: list[Action], type_: str | None = None) -> set[str]: + return {a.code for a in actions if type_ is None or a.type == type_} + + +# ---------------------------------------------------------------------------------------------- config + + +def test_yaml_loads_and_every_rule_has_evaluator() -> None: + cfg = load_rules() + ids = {r.id for r in cfg.rules} + assert ids == registered_rule_ids() + + +def test_unknown_rule_id_rejected() -> None: + with pytest.raises(ValidationError): + RulesConfig.model_validate( + {"rules": [{"id": "nope", "actions": [{"type": "alert", "level": "info"}]}]} + ) + + +def test_alert_without_level_rejected() -> None: + with pytest.raises(ValidationError): + RulesConfig.model_validate({"rules": [{"id": "sleep_lt4_3n", "actions": [{"type": "alert"}]}]}) + + +def test_disabled_rule_is_skipped() -> None: + cfg = load_rules().model_copy(deep=True) + for r in cfg.rules: + r.enabled = r.id != "sleep_lt4_3n" + s = snap(3, checkins=[ci(d, sleep_hours_bucket="<4") for d in (1, 2, 3)]) + assert "sleep_lt4_3n" not in codes(evaluate(s, cfg)) + + +def test_naive_as_of_rejected() -> None: + with pytest.raises(ValidationError): + snap(1, as_of=datetime(2026, 9, 22, 21)) + + +def test_quiet_patient_has_no_actions() -> None: + assert evaluate(snap(3, checkins=[ci(d, mood=6, med_taken=True) for d in (1, 2, 3)])) == [] + + +# ---------------------------------------------------------------------------------------------- crisis rules + + +def test_phq9_item9_crisis() -> None: + s = snap(13, scale_results=[ScalePoint(scale="phq9", ts=at(13, 20), total=12, item9=1, scheduled=False)]) + acts = evaluate(s) + crisis = [a for a in acts if a.code == "phq9_item9"] + assert [a.type for a in crisis] == ["crisis", "show_safety_plan", "schedule_scale"] + assert crisis[0].level == AlertLevel.ACUTE + assert crisis[2].params["scale"] == "cssrs_screen" + assert "1" in crisis[0].reason_ru + + +def test_phq9_item9_zero_or_old_is_not_crisis() -> None: + zero = snap(7, scale_results=[ScalePoint(scale="phq9", ts=at(7, 20), total=13, item9=0)]) + assert "phq9_item9" not in codes(evaluate(zero)) + old = snap(9, scale_results=[ScalePoint(scale="phq9", ts=at(7, 20), total=13, item9=2)]) + assert "phq9_item9" not in codes(evaluate(old)) + + +def test_crisis_never_deduplicated() -> None: + s = snap( + 13, + scale_results=[ScalePoint(scale="phq9", ts=at(13, 20), total=12, item9=2)], + open_alert_codes={"phq9_item9"}, + ) + assert "phq9_item9" in codes(evaluate(s), "crisis") + + +def test_cssrs_levels() -> None: + high = evaluate(snap(5, cssrs_risk="high")) + assert codes(high, "crisis") == {"cssrs_high"} + moderate = evaluate(snap(5, cssrs_risk="moderate")) + assert [(a.code, a.level) for a in moderate] == [("cssrs_moderate", AlertLevel.HIGH)] + assert evaluate(snap(5, cssrs_risk="low")) == [] + + +@pytest.mark.parametrize( + ("weapon", "cssrs", "item9", "fires"), + [ + (True, "moderate", None, True), + (True, None, 1, True), + (True, "low", 0, False), + (False, "high", 2, False), + (None, "moderate", None, False), + ], +) +def test_weapon_access_with_risk( + weapon: bool | None, cssrs: str | None, item9: int | None, fires: bool +) -> None: + scales = [] if item9 is None else [ScalePoint(scale="phq9", ts=at(5, 20), total=10, item9=item9)] + s = snap(5, weapon_access=weapon, cssrs_risk=cssrs, scale_results=scales) + assert ("weapon_access_risk" in codes(evaluate(s), "crisis")) is fires + + +# ---------------------------------------------------------------------------------------------- scale rules + + +@pytest.mark.parametrize( + ("scale", "total", "code", "fires"), + [ + ("phq9", 20, "phq9_severe", True), + ("phq9", 19, "phq9_severe", False), + ("pcl5", 50, "pcl5_severe", True), + ("pcl5", 49, "pcl5_severe", False), + ], +) +def test_severe_totals(scale: str, total: int, code: str, fires: bool) -> None: + s = snap(7, scale_results=[ScalePoint(scale=scale, ts=at(7, 10), total=total, item9=0)]) + acts = evaluate(s) + assert (code in codes(acts)) is fires + if fires: + assert next(a for a in acts if a.code == code).level == AlertLevel.HIGH + + +@pytest.mark.parametrize( + ("scale", "prev", "last", "code", "fires"), + [ + ("phq9", 10, 15, "phq9_rise", True), + ("phq9", 10, 14, "phq9_rise", False), + ("pcl5", 30, 40, "pcl5_rise", True), + ("pcl5", 30, 39, "pcl5_rise", False), + ("pcl5", 40, 30, "pcl5_rise", False), + ], +) +def test_rise_between_last_two(scale: str, prev: int, last: int, code: str, fires: bool) -> None: + s = snap( + 14, + scale_results=[ + ScalePoint(scale=scale, ts=at(14, 10), total=last, item9=0), # order in input does not matter + ScalePoint(scale=scale, ts=at(0, 10), total=prev, item9=0), + ], + ) + assert (code in codes(evaluate(s))) is fires + + +def test_future_scale_results_ignored() -> None: + s = snap(7, scale_results=[ScalePoint(scale="phq9", ts=at(8), total=25, item9=3)]) + assert evaluate(s) == [] + + +@pytest.mark.parametrize( + ("sex", "total", "fires"), [("m", 4, True), ("m", 3, False), ("f", 3, True), ("f", 2, False)] +) +def test_auditc_by_sex(sex: str, total: int, fires: bool) -> None: + s = snap(1, sex=sex, scale_results=[ScalePoint(scale="auditc", ts=at(0), total=total)]) + acts = evaluate(s) + assert ("auditc_positive" in codes(acts)) is fires + if fires: + assert acts[0].level == AlertLevel.MODERATE + + +# ---------------------------------------------------------------------------------------------- adherence + + +def test_quit_intent_voiced() -> None: + acts = evaluate(snap(5, checkins=[ci(5, med_taken=False, quit_intent=True)])) + assert [(a.type, a.level) for a in acts if a.code == "quit_intent"] == [ + ("alert", AlertLevel.HIGH), + ("med_info", None), + ] + assert next(a for a in acts if a.type == "med_info").params["topic"] == "delayed_effect" + + +def test_quit_intent_followed_by_skipped_dose() -> None: + s = snap( + 7, checkins=[ci(5, med_taken=True, quit_intent=True), ci(6, med_taken=True), ci(7, med_taken=False)] + ) + assert "quit_intent" in codes(evaluate(s), "alert") + no_intent = snap(7, checkins=[ci(6, med_taken=True), ci(7, med_taken=False)]) + assert "quit_intent" not in codes(evaluate(no_intent)) + + +def test_missed_doses_3_of_7() -> None: + doses = [ + DoseRecord(planned_at=at(d, 9), status=st) + for d, st in [(1, "missed"), (2, "taken"), (3, "skipped"), (4, "taken"), (5, "taken"), (6, "taken")] + ] + s = snap(7, doses=doses, checkins=[ci(7, med_taken=False)]) + acts = [a for a in evaluate(s) if a.code == "missed_doses_7d"] + assert [a.type for a in acts] == ["alert", "reach_out"] + assert acts[0].level == AlertLevel.MODERATE and acts[0].params["count"] == 3 + # Day 1 slides out of the 7-day window on day 8 → only 2 missed days left. + assert "missed_doses_7d" not in codes(evaluate(snap(8, doses=doses, checkins=[ci(7, med_taken=False)]))) + + +def test_side_effects_three_consecutive_days() -> None: + two = snap(3, checkins=[ci(1), ci(2, side_effects=["nausea"]), ci(3, side_effects=["nausea"])]) + assert evaluate(two) == [] + gap = snap( + 4, + checkins=[ + ci(1, side_effects=["nausea"]), + ci(3, side_effects=["nausea"]), + ci(4, side_effects=["nausea"]), + ], + ) + assert evaluate(gap) == [] + three = snap(4, checkins=[ci(d, side_effects=["nausea"]) for d in (2, 3, 4)]) + acts = evaluate(three) + assert [(a.code, a.level) for a in acts] == [("side_effect_3d", AlertLevel.MODERATE)] + assert "nausea" in acts[0].reason_ru + + +def test_streak_seen_in_morning_before_todays_checkin() -> None: + s = snap(5, as_of=at(5, 8), checkins=[ci(d, side_effects=["nausea"]) for d in (2, 3, 4)]) + assert "side_effect_3d" in codes(evaluate(s)) + stale = snap(6, as_of=at(6, 8), checkins=[ci(d, side_effects=["nausea"]) for d in (2, 3, 4)]) + assert evaluate(stale) == [] + + +# ---------------------------------------------------------------------------------- check-in patterns + + +def test_sleep_lt4_three_nights() -> None: + s = snap(10, checkins=[ci(d, sleep_hours_bucket="<4") for d in (8, 9, 10)]) + assert [(a.code, a.level) for a in evaluate(s)] == [("sleep_lt4_3n", AlertLevel.MODERATE)] + s2 = snap( + 10, + checkins=[ + ci(8, sleep_hours_bucket="<4"), + ci(9, sleep_hours_bucket="4-6"), + ci(10, sleep_hours_bucket="<4"), + ], + ) + assert evaluate(s2) == [] + + +def test_alcohol_three_days() -> None: + s = snap(10, checkins=[ci(8, alcohol="1-2"), ci(9, alcohol="more"), ci(10, alcohol="more")]) + acts = evaluate(s) + assert [a.type for a in acts] == ["alert", "med_info", "suggest_visit"] + assert acts[0].level == AlertLevel.MODERATE + assert acts[1].params["topic"] == "alcohol_interaction" + s2 = snap(10, checkins=[ci(8, alcohol="1-2"), ci(9, alcohol="none"), ci(10, alcohol="more")]) + assert evaluate(s2) == [] + + +def test_low_mood_schedules_unscheduled_phq9_with_cooldown() -> None: + checkins = [ci(11, mood=3), ci(12, mood=3), ci(13, mood=2)] + acts = evaluate(snap(13, checkins=checkins)) + assert [(a.type, a.level, a.params) for a in acts] == [ + ("schedule_scale", None, {"scale": "phq9", "scheduled": False}) + ] + done = ScalePoint(scale="phq9", ts=at(13, 20), total=12, item9=0, scheduled=False) + assert evaluate(snap(13, checkins=checkins, scale_results=[done])) == [] + assert evaluate(snap(13, checkins=[ci(11, mood=3), ci(12, mood=4), ci(13, mood=2)])) == [] + + +def test_silence_over_3_days() -> None: + acts = evaluate(snap(10, last_patient_message_at=at(6, 20))) + assert [(a.type, a.level) for a in acts] == [("alert", AlertLevel.INFO), ("reach_out", None)] + assert acts[0].params["silent_days"] == 4 + assert evaluate(snap(10, last_patient_message_at=at(7, 21))) == [] # exactly 3 days + + +@pytest.mark.parametrize( + ("age", "day", "fires"), [(23, 8, True), (24, 28, True), (25, 8, False), (23, 29, False)] +) +def test_young_adult_enhanced_monitoring(age: int, day: int, fires: bool) -> None: + acts = evaluate(snap(day, age=age)) + assert (acts == [a for a in acts if a.type == "enhanced_monitoring"] and bool(acts)) is fires + if fires: + assert acts[0].level == AlertLevel.INFO + + +# ---------------------------------------------------------------------------------------------- dedup + + +def test_open_alert_suppresses_whole_rule() -> None: + checkins = [ci(8, alcohol="1-2"), ci(9, alcohol="more"), ci(10, alcohol="more")] + assert evaluate(snap(10, checkins=checkins, open_alert_codes={"alcohol_3d"})) == [] + # Other rules are not affected by an unrelated open code. + assert codes(evaluate(snap(10, checkins=checkins, open_alert_codes={"sleep_lt4_3n"}))) == {"alcohol_3d"} + + +def test_evaluate_is_deterministic() -> None: + s = snap(10, checkins=[ci(d, alcohol="more", sleep_hours_bucket="<4") for d in (8, 9, 10)]) + assert evaluate(s) == evaluate(s) + + +# ---------------------------------------------------------------------------------------------- colours + + +@pytest.mark.parametrize( + ("levels", "expected"), + [ + ([], "green"), + ([AlertLevel.INFO], "grey"), + ([AlertLevel.INFO, AlertLevel.MODERATE], "yellow"), + ([AlertLevel.HIGH], "yellow"), + ([AlertLevel.MODERATE, AlertLevel.ACUTE], "red"), + ], +) +def test_color(levels: list[AlertLevel], expected: str) -> None: + assert color(levels) == expected + + +def test_patient_status() -> None: + st = patient_status([AlertLevel.MODERATE, None, AlertLevel.HIGH]) + assert st.color == "yellow" and st.top_level == AlertLevel.HIGH + assert patient_status([]).top_level is None + + +# ---------------------------------------------------------------------------------------------- demo persona +# docs/product/01_demo_persona.md, scenario B: Alexey, 28, sertraline 50 mg from day 0. +# (day, med_taken, mood, sleep, alcohol, side effects, quit_intent) +PERSONA_DAYS: list[tuple[int, bool, int, str, str, list[str], bool]] = [ + (1, True, 4, "6-8", "none", [], False), + (2, True, 4, "4-6", "none", ["nausea"], False), + (3, True, 4, "4-6", "none", ["nausea"], False), + (4, True, 3, "4-6", "none", ["nausea"], False), + (5, False, 3, "4-6", "none", [], True), # «от таблеток только хуже, брошу» + (6, True, 4, "6-8", "none", [], False), # doctor called, patient continues + (7, True, 4, "6-8", "none", [], False), + (8, True, 4, "<4", "1-2", [], False), + (9, True, 4, "<4", "more", [], False), + (10, True, 4, "<4", "more", [], False), # «не могу уснуть без пива» + (11, True, 3, "4-6", "1-2", [], False), + (12, True, 3, "4-6", "none", [], False), + (13, True, 2, "<4", "none", [], False), # «всё бессмысленно, устал от всего» +] +PERSONA_SCALES = [ + ScalePoint(scale="phq9", ts=at(-3, 12), total=14, item9=0, scheduled=True), # screening + ScalePoint(scale="phq9", ts=at(7, 19), total=13, item9=0, scheduled=True), # weekly PHQ-9 + ScalePoint(scale="phq9", ts=at(13, 20), total=18, item9=2, scheduled=False), # unscheduled (mood ≤ 3 × 3) +] +EXPECTED = ["G", "G", "G", "Y", "Y", "G", "G", "G", "G", "Y", "Y", "Y", "R"] +COLOR_LETTER = {"green": "G", "yellow": "Y", "red": "R", "grey": "g"} +DOCTOR_ACK_DAY = 6 # the doctor called the patient and acked the open alerts + + +def _persona_snapshot(day: int, open_codes: set[str], include_scales_until: datetime) -> PatientSnapshot: + rows = [r for r in PERSONA_DAYS if r[0] <= day] + return snap( + day, + checkins=[ + ci( + d, + med_taken=med, + mood=mood, + sleep_hours_bucket=sleep, + alcohol=alc, + side_effects=se, + quit_intent=qi, + ) + for d, med, mood, sleep, alc, se, qi in rows + ], + doses=[DoseRecord(planned_at=at(d, 9), status="taken" if med else "skipped") for d, med, *_ in rows], + scale_results=[p for p in PERSONA_SCALES if p.ts <= include_scales_until], + open_alert_codes=set(open_codes), + ) + + +def test_demo_persona_trajectory() -> None: + open_alerts: dict[str, AlertLevel] = {} + colours: list[str] = [] + new_alerts: dict[int, set[str]] = {} + for day in range(1, 14): + if day == DOCTOR_ACK_DAY: + open_alerts.clear() # quit_intent + side_effect_3d acked after the call + if day == 13: + # Before the unscheduled PHQ-9: mood ≤ 3 for 3 days triggers it. + pre = evaluate(_persona_snapshot(day, set(open_alerts), include_scales_until=at(13, 19))) + assert codes(pre, "schedule_scale") == {"mood_low_3d"} + acts = evaluate(_persona_snapshot(day, set(open_alerts), include_scales_until=at(day, 23))) + fresh = {a.code: a.level for a in acts if a.creates_alert and a.level is not None} + new_alerts[day] = set(fresh) + open_alerts.update(fresh) + colours.append(COLOR_LETTER[color(open_alerts.values())]) + + assert colours == EXPECTED + assert new_alerts[4] == {"side_effect_3d"} + assert new_alerts[5] == {"quit_intent"} + assert new_alerts[10] == {"sleep_lt4_3n", "alcohol_3d"} + assert new_alerts[11] == set() and new_alerts[12] == set() # still open, not re-emitted + assert new_alerts[13] == {"phq9_item9", "phq9_rise"} + assert {d for d, c in new_alerts.items() if c} == {4, 5, 10, 13} -- 2.54.0