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

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("<id>")`. 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: `<config_dir>/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))

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("<id>")`). 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 недели терапии: усиленный мониторинг"

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))

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}