"""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. Deliveries are at least once: the channel protocol offers no idempotency key. """ import asyncio import logging from contextlib import suppress from datetime import datetime from zoneinfo import ZoneInfo from sqlalchemy import select, text from sqlalchemy.exc import SQLAlchemyError from app.core import clock from app.core.intake_questions import DIARY_KINDS from app.core.intake_runtime import due_scales, intake_enabled, load_runtime from app.core.questionnaire import load_policy from app.db import repo from app.db.clinical import PatientProfile from app.db.intake import QuestionnaireTask from app.db.session import get_clinical_session from app.dialog.contracts import Button, Event, Reply, get_engine from app.dialog.texts import load_dialog_texts from app.integrations.base import integrations from app.scheduler import demo from app.services import monitoring from app.services import questionnaire as service from app.services.followup import next_day_check_due log = logging.getLogger(__name__) _worker = None _lock = asyncio.Lock() async def plan(): now = clock.now() p = load_runtime() local = now.astimezone(ZoneInfo(p.timezone)) async with get_clinical_session() as session: profiles = (await session.scalars(select(PatientProfile))).all() for profile in profiles: state = await repo.get_dialog_state(session, profile.token) if not state or state.data.get("consent") is not True or state.data.get("adult") is not True: continue crisis = state.data.get("crisis") if next_day_check_due(crisis, now): # The day after a crisis the app asks how the person is, the safety plan one tap away. texts = load_dialog_texts() plan_button = Button(label=texts.safety_plan.title.rstrip(":"), data="safety_plan") await integrations.outbox.push( profile.token, [Reply(text=texts.reach_out.after_crisis, buttons=[[plan_button]], kind="question")] ) state.data = {**state.data, "crisis": {**crisis, "followup_sent": now.isoformat()}} rows = await service.records(session, profile.token) initial = next((r for r in reversed(rows) if r.kind == "intake" and r.status == "complete"), None) if initial is None: continue diaries = [r.day for r in rows if r.kind in DIARY_KINDS and r.status == "complete"] last_activity = max([*diaries, initial.day]) high_risk = profile.enhanced_monitoring or bool(state.data.get("crisis")) if high_risk and (local.date() - last_activity).days >= p.high_risk_silence_days: await service.alert(session, profile.token, f"silence:{last_activity}", "high_risk_silence", now) if (state.data.get("crisis") or {}).get("open") or state.data.get("intake_review_required"): continue done = {r.kind for r in rows if r.day == local.date() and r.status == "complete"} # Stage 2 diary: morning block after waking, evening block at evening_hour («Этап 2», p.2). for kind, hour in (("diary_morning", p.morning_hour), ("diary_evening", p.evening_hour)): if local.hour >= hour and kind not in done: await service.queue_task(session, profile.token, kind, f"{kind}:{local.date()}", now) last = {} for row in rows: if row.status != "complete": continue for sid in ("phq9", "pcl5", "rus_audit_s"): if row.details.get("totals", {}).get(sid, {}).get("complete"): last[sid] = row.completed_at.astimezone(ZoneInfo(p.timezone)).date() route = initial.details["decision"]["primary_route"] prescriptions = await integrations.prescriptions.get_prescriptions(profile.token) changes = [ v.start_date for v in prescriptions if v.start_date and v.start_date <= local.date() and (v.end_date is None or v.end_date >= local.date()) ] if profile.therapy_start and profile.therapy_start <= local.date(): changes.append(profile.therapy_start) closed = (state.data.get("crisis") or {}).get("closed_at") crisis_date = datetime.fromisoformat(closed).astimezone(ZoneInfo(p.timezone)).date() if closed else None scales = due_scales( today=local.date(), enrolled=initial.day, last=last, route="self_help" if route == "self_help" else "specialist", medication_changed=max(changes) if changes else None, after_crisis=crisis_date, policy=p, ) # Suicide-risk module: C-SSRS «since last contact» every day, only under risk monitoring. if profile.enhanced_monitoring and local.hour >= p.evening_hour and "cssrs_since_contact" not in done: await service.queue_task(session, profile.token, "cssrs_since_contact", f"cssrs:{local.date()}", now) totals = initial.details.get("totals", {}) modules = set(profile.modules or []) policy = load_policy() # PHQ-9 only with depression or a positive PHQ-2; PCL-5 only with PTSD or a positive TSQ («Этап 2», p.3). wanted = { "phq9": "depression" in modules or bool(totals.get("phq9", {}).get("complete")), "pcl5": "veteran" in modules or (totals.get("tsq", {}).get("total") or 0) >= policy.tsq_expand_at, "rus_audit_s": True, } for sid in scales: if wanted[sid]: await service.queue_task(session, profile.token, sid, f"{sid}:{local.date()}", now) async def deliver(): async with get_clinical_session() as session: task_ids = list( ( await session.scalars( select(QuestionnaireTask.id) .where( QuestionnaireTask.sent_at.is_(None), QuestionnaireTask.status != "cancelled", QuestionnaireTask.due_at <= clock.now(), ) .order_by((QuestionnaireTask.kind != "cssrs_since_contact"), QuestionnaireTask.id) ) ).all() ) for task_id in task_ids: async with get_clinical_session() as session: task = await session.get(QuestionnaireTask, task_id) state = await repo.get_dialog_state(session, task.token) if not state or state.data.get("consent") is not True or state.data.get("adult") is not True: task.status, task.delivery = "cancelled", [] continue if (state.data.get("crisis") or {}).get("open"): continue if task.status == "done" and not task.delivery: continue token = task.token payload = task.delivery if not payload and state.state not in ("idle", "intake.collect"): continue if not payload: await get_engine().handle( token, Event(type="system", system="daily_checkin", params={"questionnaire_task": task_id}) ) # Serialize delivery with withdrawal of consent and user answers. async with get_clinical_session() as session: await service.lock_patient(session, token) state = await repo.get_dialog_state(session, token) task = await session.get(QuestionnaireTask, task_id) if not state or state.data.get("consent") is not True or not task.delivery: continue try: await asyncio.wait_for( integrations.outbox.push(token, [Reply.model_validate(r) for r in task.delivery]), timeout=10 ) except Exception as exc: log.warning("questionnaire outbox error=%s", type(exc).__name__) else: task.sent_at = clock.now() async def run_until_now(): async with _lock, get_clinical_session() as guard: if guard.bind.dialect.name == "postgresql" and not await guard.scalar( text("SELECT pg_try_advisory_xact_lock(520053)") ): 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() await service.dispatch_notices() async def _run(): while True: try: await run_until_now() except Exception as exc: log.error("questionnaire scheduler error=%s", type(exc).__name__) await asyncio.sleep(demo.load_config().tick_seconds) async def start(): global _worker try: 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() or demo.load_config().enabled): _worker = asyncio.create_task(_run()) async def stop(): global _worker if _worker is not None: _worker.cancel() with suppress(asyncio.CancelledError): await _worker _worker = None