feat(backend): demo scheduler on the virtual clock — doses, missed, check-ins, scales, silence

- app/scheduler/demo.py: run_until(now) replays due and fresh events in time order: dose rows +
  med_reminder, missed after 2 h without an answer → monitoring, daily_checkin, scale_due by the
  config/modules.yaml schedules, daily monitoring pass (silence, missed doses) → reach_out/scale_due
- idempotent via scheduler_events (unique token+key); stale events are not sent; no contact without
  consent or during an open crisis
- config/scheduler.yaml: times and windows; runs while the questionnaire release is off
- POST /demo/clock/advance replays the scheduler up to the new time

Refs #13

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

app/scheduler/demo.py

@@ -0,0 +1,257 @@
+"""Demo scheduler on the virtual clock (#13, SPEC §8, ADR-005).
+
+`run_until(now)` replays, in time order, every event that is due by `now` and still fresh
+(`config/scheduler.yaml`):
+
+  * med_reminder  — a `doses` row (pending) per prescription time, then `system:med_reminder {dose_id, drug, dose}`;
+  * missed        — a reminded dose still pending `missed_after_hours` later → status `missed` → monitoring;
+  * daily_checkin — `system:daily_checkin` at the check-in time;
+  * scale_due     — scales due by the module schedules (`config/modules.yaml`) → `system:scale_due`;
+  * evaluation    — once a day `MonitoringService.after_data_change()`: catches silence > N days (rule
+                    `no_contact_3d`) and missed doses even when the patient writes nothing.
+
+The rules decide (ADR-007); the scheduler only turns their dialog actions into system events for the engine
+(`reach_out` → `system:reach_out`, `schedule_scale` → `system:scale_due`). Replies go to `integrations.outbox`.
+
+Idempotent: a delivered event is stored in `scheduler_events` under a unique key and never sent again; a failed
+delivery is not stored and is retried on the next run while it is fresh (at least once). Only patients with
+consent (and adult) are contacted; while a crisis is open no reminders, check-ins or scales are sent.
+Time comes only from the arguments / `app.core.clock` (guardrail #4).
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+from datetime import date, datetime, time, timedelta
+from functools import lru_cache
+from pathlib import Path
+from typing import Any
+
+import yaml
+from pydantic import BaseModel, ConfigDict, Field
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.config import get_settings
+from app.core.modules import due_scales, load_modules
+from app.core.rules import Action
+from app.db import repo
+from app.db.clinical import DialogState, Dose, PatientProfile, Prescription
+from app.db.scheduler import SchedulerEvent
+from app.db.session import get_clinical_session
+from app.dialog.contracts import Event, Reply, get_engine
+from app.integrations.base import integrations
+from app.services.contracts import get_monitoring
+
+log = logging.getLogger(__name__)
+
+# ------------------------------------------------------------------------------------------------ config
+
+
+class _Strict(BaseModel):
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+
+class DosesConfig(_Strict):
+    late_hours: float = Field(gt=0)
+    missed_after_hours: float = Field(gt=0)
+
+
+class Window(_Strict):
+    time: time
+    late_hours: float = Field(gt=0)
+
+
+class SchedulerConfig(_Strict):
+    enabled: bool = True
+    tick_seconds: float = Field(gt=0)
+    doses: DosesConfig
+    daily_checkin: Window
+    scales: Window
+    evaluation: Window
+
+
+@lru_cache
+def load_config(path: Path | None = None) -> SchedulerConfig:
+    path = path or get_settings().config_dir / "scheduler.yaml"
+    return SchedulerConfig.model_validate(yaml.safe_load(path.read_text(encoding="utf-8")))
+
+
+# ------------------------------------------------------------------------------------------------ jobs
+
+# Same due time → this order: the dose before the check-in that asks about it.
+_ORDER = {"evaluation": 0, "missed": 1, "med_reminder": 2, "scale_due": 3, "daily_checkin": 4}
+
+
+@dataclass(order=True)
+class Job:
+    due: datetime
+    rank: int
+    token: str
+    key: str
+    kind: str = field(compare=False)
+    params: dict[str, Any] = field(default_factory=dict, compare=False)
+
+
+def _job(due: datetime, token: str, key: str, kind: str, **params: Any) -> Job:
+    return Job(due=due, rank=_ORDER[kind], token=token, key=key, kind=kind, params=params)
+
+
+def _at(day: date, t: time, now: datetime) -> datetime:
+    return datetime.combine(day, t, tzinfo=now.tzinfo)
+
+
+def _fresh(due: datetime, now: datetime, late_hours: float) -> bool:
+    return due <= now < due + timedelta(hours=late_hours)
+
+
+def _window(day: date, w: Window, now: datetime) -> datetime | None:
+    due = _at(day, w.time, now)
+    return due if _fresh(due, now, w.late_hours) else None
+
+
+async def _dose_jobs(s: AsyncSession, p: PatientProfile, now: datetime, cfg: SchedulerConfig) -> list[Job]:
+    jobs: list[Job] = []
+    today = now.date()
+    for rx in (await s.scalars(select(Prescription).where(Prescription.token == p.token))).all():
+        start = rx.start_date or p.therapy_start
+        for day in (today - timedelta(days=1), today):  # a fresh dose is never older than a day
+            if (start and day < start) or (rx.end_date and day > rx.end_date):
+                continue
+            for raw in (rx.schedule or {}).get("times", []):
+                planned = _at(day, time.fromisoformat(raw), now)
+                if _fresh(planned, now, cfg.doses.late_hours):
+                    key = f"dose:{rx.id}:{planned.isoformat(timespec='minutes')}"
+                    jobs.append(_job(planned, p.token, key, "med_reminder", rx_id=rx.id, planned_at=planned))
+    return jobs
+
+
+async def _missed_jobs(s: AsyncSession, token: str, now: datetime, cfg: SchedulerConfig) -> list[Job]:
+    after = timedelta(hours=cfg.doses.missed_after_hours)
+    stmt = select(Dose).where(Dose.token == token, Dose.status == "pending", Dose.planned_at <= now - after)
+    return [_job(d.planned_at + after, token, f"missed:{d.id}", "missed", dose_id=d.id) for d in await s.scalars(stmt)]
+
+
+async def _scale_jobs(s: AsyncSession, p: PatientProfile, now: datetime, cfg: SchedulerConfig) -> list[Job]:
+    due_at = _window(now.date(), cfg.scales, now)
+    if due_at is None:
+        return []
+    last: dict[str, date] = {}
+    for r in await repo.list_scale_results(s, p.token):
+        if r.complete:
+            last[r.scale] = r.ts.astimezone(now.tzinfo).date()
+    if p.therapy_start:
+        # A scale never taken counts from the visit where therapy started, not as overdue on day 0.
+        for mod in load_modules().modules.values():
+            for entry in mod.schedule:
+                last.setdefault(entry.scale, p.therapy_start)
+    therapy_day = (now.date() - p.therapy_start).days if p.therapy_start else 0
+    scales = due_scales(set(p.modules or []), therapy_day, last, now.date())
+    return [_job(due_at, p.token, f"scale:{sc}:{now.date()}", "scale_due", scale=sc) for sc in scales]
+
+
+async def collect(now: datetime, cfg: SchedulerConfig | None = None) -> list[Job]:
+    """Due and fresh, not yet delivered events of all contactable patients, in time order."""
+    cfg = cfg or load_config()
+    jobs: list[Job] = []
+    async with get_clinical_session() as s:
+        for p in (await s.scalars(select(PatientProfile).order_by(PatientProfile.token))).all():
+            state = await s.get(DialogState, p.token)
+            data = state.data if state is not None else {}
+            if data.get("consent") is not True or data.get("adult") is not True:
+                continue
+            mine = await _missed_jobs(s, p.token, now, cfg)
+            if (due := _window(now.date(), cfg.evaluation, now)) is not None:
+                mine.append(_job(due, p.token, f"eval:{now.date()}", "evaluation"))
+            if not (data.get("crisis") or {}).get("open"):
+                mine += await _dose_jobs(s, p, now, cfg)
+                mine += await _scale_jobs(s, p, now, cfg)
+                if (due := _window(now.date(), cfg.daily_checkin, now)) is not None:
+                    mine.append(_job(due, p.token, f"checkin:{now.date()}", "daily_checkin"))
+            done = set((await s.scalars(select(SchedulerEvent.key).where(SchedulerEvent.token == p.token))).all())
+            jobs += [j for j in mine if j.key not in done]
+    return sorted(jobs)
+
+
+# ------------------------------------------------------------------------------------------------ delivery
+
+
+async def _system(token: str, kind: str, **params: Any) -> None:
+    replies: list[Reply] = await get_engine().handle(
+        token,
+        Event(type="system", system=kind, params=params),  # type: ignore[arg-type]
+    )
+    if replies:
+        await integrations.outbox.push(token, replies)
+
+
+async def _after_rules(token: str) -> None:
+    """Monitoring pass outside a dialog: hand the rules' dialog actions to the engine as system events."""
+    actions: list[Action] = await get_monitoring().after_data_change(token)
+    for a in actions:
+        if a.type == "reach_out":
+            await _system(token, "reach_out", **a.params)
+        elif a.type == "schedule_scale" and a.params.get("scale"):
+            await _system(token, "scale_due", scale=a.params["scale"], scheduled=bool(a.params.get("scheduled")))
+        elif a.type == "crisis":
+            log.warning("scheduler crisis action token=%s code=%s", token, a.code)  # alert already persisted
+
+
+async def _ensure_dose(token: str, rx_id: int, planned_at: datetime) -> tuple[int, Prescription]:
+    async with get_clinical_session() as s:
+        rx = await s.get(Prescription, rx_id)
+        assert rx is not None
+        dose = await s.scalar(
+            select(Dose).where(Dose.token == token, Dose.prescription_id == rx_id, Dose.planned_at == planned_at)
+        )
+        if dose is None:
+            dose = Dose(token=token, prescription_id=rx_id, planned_at=planned_at, status="pending")
+            s.add(dose)
+            await s.flush()
+        return dose.id, rx
+
+
+async def _mark_missed(dose_id: int) -> bool:
+    async with get_clinical_session() as s:
+        dose = await s.get(Dose, dose_id)
+        if dose is None or dose.status != "pending":
+            return False  # answered meanwhile
+        dose.status = "missed"
+        return True
+
+
+async def _run(job: Job, now: datetime) -> None:
+    if job.kind == "med_reminder":
+        dose_id, rx = await _ensure_dose(job.token, job.params["rx_id"], job.params["planned_at"])
+        await _system(job.token, "med_reminder", dose_id=dose_id, drug=rx.drug, dose=rx.dose)
+    elif job.kind == "missed":
+        if await _mark_missed(job.params["dose_id"]):
+            await _after_rules(job.token)
+    elif job.kind == "daily_checkin":
+        await _system(job.token, "daily_checkin")
+    elif job.kind == "scale_due":
+        await _system(job.token, "scale_due", scale=job.params["scale"], scheduled=True)
+    elif job.kind == "evaluation":
+        await _after_rules(job.token)
+
+
+async def _done(job: Job, now: datetime) -> None:
+    async with get_clinical_session() as s:
+        s.add(SchedulerEvent(token=job.token, key=job.key, kind=job.kind, due_at=job.due, done_at=now))
+
+
+async def run_until(now: datetime, cfg: SchedulerConfig | None = None) -> list[Job]:
+    """Deliver every due event up to `now` in time order; returns the delivered jobs."""
+    delivered: list[Job] = []
+    for job in await collect(now, cfg):
+        try:
+            await _run(job, now)
+        except Exception as exc:  # noqa: BLE001 — one patient's failure must not stop the others
+            log.error("scheduler job failed token=%s kind=%s error=%s", job.token, job.kind, type(exc).__name__)
+            continue
+        await _done(job, now)
+        delivered.append(job)
+    if delivered:
+        log.info("scheduler delivered=%s", [(j.token, j.kind) for j in delivered])
+    return delivered

app/scheduler/jobs.py

@@ -1,4 +1,4 @@
-"""Questionnaire jobs on the demo clock; clinical release disabled by default.
+"""Scheduler on the demo clock: questionnaire jobs when that release is on, else the demo scheduler (#13).
 
 One scheduler per database. PostgreSQL's advisory transaction lock serializes
 workers; an asyncio lock does the same within a SQLite demonstration process.
@@ -22,6 +22,7 @@ from app.db.intake import QuestionnaireTask
 from app.db.session import get_clinical_session
 from app.dialog.contracts import Event, Reply, get_engine
 from app.integrations.base import integrations
+from app.scheduler import demo
 from app.services import monitoring
 from app.services import questionnaire as service
 
@@ -148,6 +149,9 @@ async def run_until_now():
             return
         await service.dispatch_notices()
         if not intake_enabled():
+            # Questionnaire release off → the demo scheduler (doses, check-ins, scales, silence) runs instead.
+            if demo.load_config().enabled:
+                await demo.run_until(clock.now())
             return
         await plan()
         await deliver()
@@ -160,7 +164,7 @@ async def _run():
             await run_until_now()
         except Exception as exc:
             log.error("questionnaire scheduler error=%s", type(exc).__name__)
-        await asyncio.sleep(30)
+        await asyncio.sleep(demo.load_config().tick_seconds)
 
 
 async def start():
@@ -169,7 +173,7 @@ async def start():
         await monitoring.init_clock_persistence()
     except (OSError, SQLAlchemyError) as exc:
         log.error("clock persistence unavailable error=%s", type(exc).__name__)
-    if _worker is None and intake_enabled():
+    if _worker is None and (intake_enabled() or demo.load_config().enabled):
         _worker = asyncio.create_task(_run())