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