Files
PTSRaiAgent/app/dialog/flows/intake.py
T
ElebobaandClaude Opus 5.5 531bef1e73
CI / ruff · pytest (pull_request) Successful in 2m33s
feat: crisis message after the test and non-blocking; weapon answer triggers nothing; ЕМИАС demo = Никита
- CRISIS_BLOCKING (default off for the demo): a crisis no longer locks the dialog; the message and the
  doctor's acute alert still happen. The clinical protocol (blocking, immediate stop) stays behind the flag and
  keeps its tests.
- Crisis signals inside a questionnaire (intake red flags, high C-SSRS, PHQ-9 item 9) no longer stop it: the
  test is finished and the crisis message comes at the end; the doctor's alert is raised at once/at the end.
- Weapon access: asked and stored, triggers nothing (rule disabled, no red flag, no safety block).
- ЕМИАС demo persona is «Никита В.»; the one-click demo stays «Андрей С.».

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-04 23:30:05 +03:00

555 lines
24 KiB
Python

"""Versioned, resumable questionnaires. Disabled until the clinical release gate opens."""
import math
from dataclasses import asdict, replace
from zoneinfo import ZoneInfo
from app.config import get_settings
from app.core import interpretation
from app.core.intake_questions import (
DIARY_KINDS,
Question,
content,
cssrs_answers,
cssrs_since_contact_question,
diary_questions,
group_answers,
next_intake,
red_flag_category,
red_flag_triggers,
scale_question,
texts,
)
from app.core.intake_runtime import diary_flags, intake_enabled, load_runtime
from app.core.questionnaire import alcohol_expansion, load_policy, score_alcohol, score_cssrs_recent
from app.core.scales import load_scale
from app.core.types import AlertLevel
from app.db import repo
from app.db.intake import QuestionnaireRecord, QuestionnaireTask
from app.dialog.contracts import Button, Event, Reply
from app.dialog.crisis import crisis_open, open_crisis
from app.dialog.fsm import Step, on_enter, state, trigger
from app.services import questionnaire as service
from app.services.followup import schedule_post_crisis_call
def allowed(ctx):
return ctx.data.get("consent") is True and ctx.data.get("adult") is True and intake_enabled()
def side_effects_question():
"""FIBSER-style tolerability question added to the full scales while a medication is active («Этап 2», p.3)."""
return Question("side_effects", texts()["side_effects"], ((1, "Да"), (0, "Нет")), "past_week")
def next_question(row):
a = row.answers
if row.kind == "intake":
return next_intake(a, load_policy(), row.details.get("sex"))
if row.kind == "cssrs_since_contact":
return cssrs_since_contact_question(a)
if row.kind in DIARY_KINDS:
part = row.kind.removeprefix("diary_")
for q in diary_questions(part, row.details["modules"], day_index=row.details.get("day_index", 0)):
if q.id in a:
continue
if q.id == "medication_reason" and a.get("medication") != "no":
continue
return q
return None
if row.kind in ("phq9", "pcl5", "rus_audit_s"):
sid = row.kind
if sid == "rus_audit_s":
short = score_alcohol(
{k: a[k] for k in load_scale(sid).item_ids if k in a}, short=True, sex=row.details.get("sex")
)
if alcohol_expansion(short, load_policy()):
sid = "rus_audit"
for item in load_scale(sid).item_ids:
if item == "rus_audit_2" and a.get("rus_audit_1") == 0:
continue
if item not in a:
q = scale_question(sid, item)
if sid == "pcl5" and row.details["weekly_pcl"]:
q = replace(q, text=q.text.replace("последний месяц", "последнюю неделю"), period="past_week")
return q
if row.details.get("medication_active") and "side_effects" not in a:
return side_effects_question()
return None
def question_reply(row, q):
t = texts()
# Nonce + item + option index rejects old/replayed callbacks without exposing raw answers.
prefix = f"iq:{row.id[:12]}:{q.id}:"
buttons = [[Button(label=label, data=prefix + str(i))] for i, (_, label) in enumerate(q.options)]
if q.optional:
buttons.append([Button(label=t["comment_skip"], data=prefix + "skip")])
buttons.extend([[Button(label=t["help"], data="intake:help")], [Button(label=t["pause"], data="intake:pause")]])
return Reply(text=q.text + "\n\n" + t["as_of"].format(day=row.day.isoformat()), buttons=buttons, kind="question")
async def urgent(ctx, row=None, *, code="help_now", categories=None):
if crisis_open(ctx.data):
return open_crisis(ctx, categories or [], source="questionnaire:" + code)
profile = await ctx.profile()
if profile:
profile.enhanced_monitoring = True
flags = list(service.decision(row).flags) if row else []
if row and (row.answers.get("dphq_9", 0) > 0 or row.answers.get("phq9_9", 0) > 0):
flags.append("cssrs_since_contact_required")
if row:
await service.finish(ctx, row, "interrupted", extra={"stop_reason": code, "flags": flags})
await service.alert(
ctx.session,
ctx.token,
f"{row.id if row else ctx.now.isoformat()}:{code}",
code,
ctx.now,
level=AlertLevel.ACUTE,
params={"record_id": row.id if row else None, "flags": flags},
)
await schedule_post_crisis_call(ctx.session, ctx.token, ctx.now, source="questionnaire:" + code)
step = open_crisis(ctx, categories or [], source="questionnaire:" + code)
step.data["questionnaire_record"] = None
step.data["intake_review_required"] = True
for reply in step.replies:
reply.buttons.append([Button(label=ctx.texts.safety_plan.title.rstrip(":"), data="safety_plan")])
return step
def _defer_urgent(row, code, categories=None):
"""Remember the first crisis signal of this questionnaire; the questionnaire goes on (team decision
2026-10-04: the crisis message comes at the end and blocks nothing)."""
if not row.details.get("pending_urgent"):
row.details = {**row.details, "pending_urgent": {"code": code, "categories": categories or []}}
async def _flush_urgent(ctx, row, step):
"""Questionnaire over: acute alert + callback for the doctor, then the (non-blocking) crisis message."""
pending = (row.details or {}).get("pending_urgent")
if not pending or step.next_state == "intake.collect":
return step
row.details = {**row.details, "pending_urgent": None}
code, categories = pending["code"], pending.get("categories") or []
profile = await ctx.profile()
if profile:
profile.enhanced_monitoring = True
await service.alert(
ctx.session,
ctx.token,
f"{row.id}:{code}",
code,
ctx.now,
level=AlertLevel.ACUTE,
params={"record_id": row.id, "deferred": True},
)
await schedule_post_crisis_call(ctx.session, ctx.token, ctx.now, source="questionnaire:" + code)
crisis = open_crisis(ctx, categories, source="questionnaire:" + code)
for reply in crisis.replies:
reply.buttons.append([Button(label=ctx.texts.safety_plan.title.rstrip(":"), data="safety_plan")])
return step.then(crisis)
async def item9_discrepancy(ctx, row):
profile = await ctx.profile()
if profile:
profile.enhanced_monitoring = True
await service.alert(
ctx.session,
ctx.token,
row.id + ":phq9_cssrs_discrepancy",
"phq9_cssrs_discrepancy",
ctx.now,
level=AlertLevel.HIGH,
params={"record_id": row.id},
)
row.details = {**row.details, "item9_flagged": True}
async def item9_assessment(ctx, row):
"""Thoughts of death in PHQ-9 item 9 → the C-SSRS screener at once; the screener's level decides (graded)."""
await service.alert(
ctx.session,
ctx.token,
row.id + ":item9_assessment",
"phq9_item9_cssrs",
ctx.now,
level=AlertLevel.HIGH,
params={"record_id": row.id},
)
await service.finish(ctx, row, "complete", extra={"stop_reason": "phq9_item9_cssrs"})
profile = await ctx.profile()
if profile:
profile.enhanced_monitoring = True
task = await service.queue_task(ctx.session, ctx.token, "cssrs_since_contact", row.id + ":cssrs", ctx.now)
return await start(ctx, "cssrs_since_contact", task)
async def manual_review(ctx, row, code):
await service.finish(ctx, row, "review_required", extra={"stop_reason": code})
await service.alert(
ctx.session, ctx.token, f"{row.id}:{code}", code, ctx.now, level=AlertLevel.HIGH, params={"record_id": row.id}
)
return Step(
replies=[
Reply(
text=texts()["missing_followup"] if code == "cssrs_since_contact_required" else texts()["review"],
buttons=[[Button(label=texts()["help"], data="intake:help")]],
kind="info",
)
],
next_state="idle",
data={"questionnaire_record": None, "intake_review_required": True},
)
async def advance(ctx, row):
return await _flush_urgent(ctx, row, await _advance(ctx, row))
async def _advance(ctx, row):
a = row.answers
c = score_cssrs_recent(cssrs_answers(a))
policy = load_policy()
# Red flags (asked first): any «Да» stops the questionnaire with the help screen of its category.
for item_id, value in a.items():
if item_id.startswith("rf_") and value == 1 and red_flag_triggers(item_id):
category = red_flag_category(item_id)
if get_settings().crisis_blocking: # clinical protocol: stop at once
return await urgent(ctx, row, code="red_flag_" + item_id[3:], categories=[category] if category else [])
_defer_urgent(row, "red_flag_" + item_id[3:], [category] if category else [])
break
# One crisis rule for the whole app (ADR-013): the crisis screen opens at a HIGH C-SSRS risk only. Moderate →
# stop + psychiatrist (about 24 h); low → flag, safety plan, continue.
if c.risk_lower_bound == "high":
if get_settings().crisis_blocking:
return await urgent(ctx, row, code="suicide_signal")
_defer_urgent(row, "suicide_signal")
if a.get("dphq_9", 0) > 0 and row.kind in DIARY_KINDS:
# Diary depression module, item 9 ≥1 → the C-SSRS right away («Этап 2»); its risk level decides.
return await item9_assessment(ctx, row)
if a.get("phq9_9", 0) > 0 and not c.complete:
# Item 9 without a completed screener (only in a scheduled PHQ-9): assess the risk right away.
return await item9_assessment(ctx, row)
if a.get("phq9_9", 0) > 0 and not row.details.get("item9_flagged"):
# Item 9 ≥1 with a negative / low C-SSRS = discrepancy (routing table): flag to the doctor, safety plan.
await item9_discrepancy(ctx, row)
if c.positive and policy.cssrs_positive_policy == "stop_any":
return await urgent(ctx, row, code="cssrs_positive")
if c.conflicts:
return await manual_review(ctx, row, "cssrs_conflict")
# Finish the risk questions before any routine question; a moderate result stops here.
if c.complete and c.risk == "moderate":
return await manual_review(ctx, row, "urgent_psychiatrist")
if c.complete and c.risk == "low":
profile = await ctx.profile()
if profile:
profile.enhanced_monitoring = True
await service.alert(
ctx.session,
ctx.token,
row.id + ":cssrs_low",
"cssrs_low_safety_plan",
ctx.now,
params={"record_id": row.id},
)
try:
q = next_question(row)
# Validate preserved short answers against a newly answered full-form item 1.
score_alcohol(group_answers(a, "rus_audit"), short=False, sex=row.details.get("sex"))
except ValueError:
return await manual_review(ctx, row, "answer_conflict")
if q:
replies = [question_reply(row, q)]
if ((c.complete and c.risk == "low") or row.details.get("item9_flagged")) and not row.details.get("plan_shown"):
from app.dialog.actions import safety_plan_reply
replies.insert(0, await safety_plan_reply(ctx))
row.details = {**row.details, "plan_shown": True}
return Step(replies=replies, next_state="intake.collect", data={"questionnaire_record": row.id})
extra = {}
if a.get("side_effects") == 1:
await service.alert(
ctx.session,
ctx.token,
row.id + ":side_effects_review",
"side_effects_review",
ctx.now,
params={"record_id": row.id},
)
if row.kind == "intake":
result = service.decision(row)
if result.status != "ready":
return await manual_review(ctx, row, "intake_incomplete_review")
extra = {"decision": asdict(result), "totals": service.summary(row)}
if a.get("recent_trauma") == 1:
extra["recent_trauma"] = True
extra["interpretation"] = interpretation.intake_interpretation(extra["totals"], a, result, policy)
await apply_modules(ctx, extra["interpretation"]["modules"])
if result.primary_route != "self_help" or result.flags or result.parallel_routes or a.get("recent_trauma"):
await service.alert(
ctx.session,
ctx.token,
row.id + ":route",
"intake_route_review",
ctx.now,
params={"record_id": row.id, **extra},
)
elif row.kind == "cssrs_since_contact":
extra["totals"] = {"cssrs_recent": asdict(c)}
extra["interpretation"] = interpretation.cssrs_interpretation(c.risk)
elif row.kind in DIARY_KINDS:
# Morning and evening answers of one date form one diary day.
history: dict = {}
for r in await service.records(ctx.session, ctx.token):
if r.kind in DIARY_KINDS and r.status == "complete" and r.id != row.id:
history.setdefault(r.day, {}).update(r.answers)
history.setdefault(row.day, {}).update(a)
flags = diary_flags(history, row.day, medication_active=row.details.get("medication_active", False))
extra["flags"] = flags
for flag in flags:
await service.alert(
ctx.session, ctx.token, row.id + ":" + flag, flag, ctx.now, params={"record_id": row.id}
)
if flag in DIARY_SCALE_TASKS:
sid = DIARY_SCALE_TASKS[flag]
await service.queue_task(ctx.session, ctx.token, sid, row.id + ":" + sid, ctx.now)
extra["interpretation"] = interpretation.diary_interpretation(row.kind.removeprefix("diary_"), flags)
else:
extra["totals"] = service.summary(row)
runtime = load_runtime()
if row.kind in ("phq9", "pcl5"):
sid = row.kind
total = extra["totals"][sid]["total"]
# Only PHQ-9 has a priority threshold; PCL-5 alone never makes an appointment a priority.
priority = getattr(policy, sid + "_priority_at", None)
specialist = getattr(policy, sid + "_specialist_at")
extra["route"] = (
"priority_psychiatrist"
if priority is not None and total >= priority
else ("psychiatrist_or_psychotherapist" if total >= specialist else "review_with_treating_clinician")
)
earlier = [
r
for r in await service.records(ctx.session, ctx.token)
if r.id != row.id
and r.status == "complete"
and r.version == row.version
and r.details.get("totals", {}).get(sid, {}).get("complete")
and (sid != "pcl5" or r.details.get("weekly_pcl", False) == row.details.get("weekly_pcl", False))
]
prior = earlier[-1] if earlier else None
if prior:
change = total - prior.details["totals"][sid]["total"]
extra["delta"] = {
"baseline_record": prior.id,
"value": change,
"retriage": change >= getattr(runtime, sid + "_retriage_delta"),
}
extra["interpretation"] = interpretation.scale_interpretation(
sid,
extra["totals"],
baseline=earlier[0].details["totals"][sid]["total"] if earlier else None,
previous=prior.details["totals"][sid]["total"] if prior else None,
policy=policy,
retriage_delta=getattr(runtime, sid + "_retriage_delta"),
)
if row.kind == "rus_audit_s" and not row.details.get("sex"):
return await manual_review(ctx, row, "alcohol_sex_unknown")
if row.kind == "rus_audit_s":
extra["interpretation"] = interpretation.scale_interpretation(
"rus_audit_s", extra["totals"], baseline=None, previous=None, policy=policy, retriage_delta=0
)
if extra["totals"]["rus_audit"].get("zone") == 4:
extra["parallel_routes"] = ["narcologist"]
retriage = extra.get("delta", {}).get("retriage", False)
await service.alert(
ctx.session,
ctx.token,
row.id + ":followup",
"followup_retriage_required" if retriage else "followup_review",
ctx.now,
level=AlertLevel.HIGH if retriage or extra.get("route") == "priority_psychiatrist" else AlertLevel.MODERATE,
params={"record_id": row.id, **extra},
)
await service.finish(ctx, row, "complete", extra=extra)
shown = interpretation.render(extra["interpretation"]) if extra.get("interpretation") else texts()["done"]
if row.kind == "cssrs_since_contact" and not await done_today(ctx, "diary_evening"):
# The risk questions open the evening diary; continue straight into it.
step = await start(ctx, "diary_evening")
step.replies.insert(0, Reply(text=shown, kind="info"))
return step
from app.dialog.flows.core import menu_reply
replies = [menu_reply(ctx, shown)]
if ((c.complete and c.risk == "low") or row.details.get("item9_flagged")) and not row.details.get("plan_shown"):
# The safety plan is never skipped because the risk signal came with the last question.
from app.dialog.actions import safety_plan_reply
replies.insert(0, await safety_plan_reply(ctx))
return Step(replies=replies, next_state="idle", data={"questionnaire_record": None})
# Diary flags that bring a full scale forward («Этап 2», p.4).
DIARY_SCALE_TASKS = {
"phq9_due_mood": "phq9",
"phq9_due_depression": "phq9",
"pcl5_due_ptsd": "pcl5",
"rus_audit_due_alcohol": "rus_audit_s",
}
async def apply_modules(ctx, modules):
"""Stage 1 result → diary modules of stage 2. Without a doctor the screening defines them; a doctor's modules
(diagnosis from ЕМИАС) are kept and only extended."""
profile = await ctx.profile()
if profile is None:
await repo.upsert_profile(ctx.session, token=ctx.token, doctor_id="", modules=list(modules))
return
current = list(profile.modules or [])
merged = [*current, *(m for m in modules if m not in current)] if profile.doctor_id else list(modules)
await repo.upsert_profile(ctx.session, token=ctx.token, modules=merged)
def local_day(ctx):
return ctx.now.astimezone(ZoneInfo(load_runtime().timezone)).date()
async def done_today(ctx, kind):
day = local_day(ctx)
return any(
r.kind == kind and r.day == day and r.status == "complete"
for r in await service.records(ctx.session, ctx.token)
)
async def start(ctx, kind="intake", task=None):
if not allowed(ctx) or crisis_open(ctx.data):
if task:
return Step()
return Step(replies=[Reply(text=texts()["unavailable"], kind="info")])
existing = await service.records(ctx.session, ctx.token)
active = next((r for r in reversed(existing) if r.status in ("active", "paused")), None)
if active:
if task and task.record_id != active.id:
return Step()
if (
active.policy_version != load_policy().version
or active.version != content()["version"]
or active.details.get("runtime_version") != load_runtime().version
):
return await manual_review(ctx, active, "questionnaire_version_changed")
active.status = "active"
return await advance(ctx, active)
if ctx.data.get("intake_review_required") and kind != "cssrs_since_contact":
if task:
return Step()
return Step(replies=[Reply(text=texts()["review"], kind="info")])
# A manual start takes over the scheduled task of the same form, so the task does not stay outstanding.
task = task or await service.outstanding_task(ctx.session, ctx.token, kind)
row = await service.create_record(ctx, kind, task=task)
return await advance(ctx, row)
@on_enter("intake.start")
async def enter(ctx):
return await start(ctx)
@trigger("button:intake:start")
@trigger("button:intake:resume")
async def start_button(ctx, event):
return await start(ctx)
@trigger("button:menu:checkin", replace=True)
@trigger("system:daily_checkin", replace=True)
async def checkin(ctx, event):
if not intake_enabled():
return Step(replies=[Reply(text=ctx.texts.idle.stub_checkin)])
if event.params.get("questionnaire_task"):
task = await ctx.session.get(QuestionnaireTask, event.params["questionnaire_task"])
if task is None or task.token != ctx.token or task.status not in ("pending", "active"):
return Step()
return await start(ctx, task.kind, task)
existing = await service.records(ctx.session, ctx.token)
if not any(r.kind == "intake" and r.status == "complete" for r in existing):
return await start(ctx)
return await next_diary(ctx)
async def next_diary(ctx):
"""The part of today's diary that is still open: morning first, then evening. The evening opens with the C-SSRS
«since last contact» when the profile is under suicide-risk monitoring («Этап 2», module of suicide risk)."""
day = local_day(ctx)
done = {r.kind for r in await service.records(ctx.session, ctx.token) if r.day == day and r.status == "complete"}
for kind in DIARY_KINDS:
if kind in done:
continue
profile = await ctx.profile()
if kind == "diary_evening" and profile and profile.enhanced_monitoring and "cssrs_since_contact" not in done:
risk_task = await service.queue_task(ctx.session, ctx.token, "cssrs_since_contact", f"cssrs:{day}", ctx.now)
return await start(ctx, "cssrs_since_contact", risk_task)
return await start(ctx, kind)
return Step(replies=[Reply(text=texts()["diary_done"], kind="info")])
@trigger("button:diary:next")
async def diary_button(ctx, event):
return await next_diary(ctx)
@state("intake")
async def collect(ctx, event: Event):
if not allowed(ctx):
return Step(replies=[Reply(text=texts()["unavailable"], kind="info")], next_state="idle")
row = await ctx.session.get(QuestionnaireRecord, ctx.data.get("questionnaire_record", ""))
if row is None or row.token != ctx.token or row.status != "active":
return Step(next_state="idle")
if (
row.policy_version != load_policy().version
or row.version != content()["version"]
or row.details.get("runtime_version") != load_runtime().version
):
return await manual_review(ctx, row, "questionnaire_version_changed")
if event.type == "system":
task_id = event.params.get("questionnaire_task")
if task_id and row.details.get("task_id") == task_id:
return await advance(ctx, row)
return Step()
if event.button == "intake:pause":
row.status = "paused"
return Step(
replies=[Reply(text=texts()["paused"], buttons=[[Button(label=texts()["resume"], data="intake:resume")]])],
next_state="idle",
)
q = next_question(row)
if q is None:
return await advance(ctx, row)
prefix = f"iq:{row.id[:12]}:{q.id}:"
value = None
valid = False
if event.type == "button" and (event.button or "").startswith(prefix):
index = event.button[len(prefix) :]
if index == "skip" and q.optional:
valid = True
elif index.isdigit() and int(index) < len(q.options):
value, valid = q.options[int(index)][0], True
elif event.type == "text" and q.numeric:
try:
value = float(str(ctx.text).replace(",", "."))
valid = math.isfinite(value) and 0 <= value <= q.maximum
except ValueError:
pass
elif event.type == "text" and q.optional:
value, valid = str(ctx.text)[:2000], True
if not valid:
hint = "invalid" if not q.numeric else ("numeric_portions" if q.id == "alcohol_portions" else "numeric")
return Step(replies=[Reply(text=texts()[hint]), question_reply(row, q)])
service.save_answer(row, q, value, ctx.now)
if row.kind == "cssrs_since_contact" and q.id == "cssrs_q6_lifetime":
# Behaviour since the last contact is recent by definition (see cssrs_since_contact_question).
service.save_answer(row, replace(q, id="cssrs_q6_recent"), value, ctx.now, derived=True)
return await advance(ctx, row)