PTSRaiAgent/app/core/intake_runtime.py

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