Arxip222andClaude Opus 5.5
44814b7293
Clinical review of the questionnaires (ADR-013):
- Red-flag block first in stage 1 (config/instruments/red_flags.yaml); any «Да» stops the
questionnaire with an acute alert and the help screen of its category.
- One crisis rule: the crisis screen opens at a HIGH C-SSRS risk only; moderate stops and goes
to a psychiatrist (24 h as a guide), low is a flag with the safety plan. PHQ-9 item 9 with a
negative screener is a discrepancy flag; without a completed screener it asks the C-SSRS at once.
config/questionnaire.yaml: cssrs_positive_policy graded (C01).
- Every crisis leaves a callback task for the doctor on duty (post_crisis_call, one per patient)
and the app asks how the person is the next day, the safety plan one tap away.
- Routing: risk and flags first, scales second; PCL-5 >= 50 no longer means a priority
appointment (pcl5_severe is moderate on the dashboard too).
- Crisis numbers live only in config/texts/crisis.yaml; texts and the agent prompt refer to them
via {phone:<key>}. Fixes the children's helpline offered to adults by the chat agent.
Refs #92
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
211 lines
9.7 KiB
Python
211 lines
9.7 KiB
Python
"""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
|