ElebobaandClaude Opus 5.5
fa40be3d94
CI / ruff · pytest (pull_request) Successful in 25s
- Engine.handle registered via register_engine; lexicon-first safety (no LLM on ACUTE), fixed crisis script once per open crisis, POSSIBLE → C-SSRS screen - FSM registry (@state/@on_enter/@trigger) with auto-loaded flows; core flow: new → consent → age_gate → module onboarding → idle, crisis, consent_revoked - reusable scale session (buttons, extract_item + clarify, item 9 / C-SSRS buttons-only, crisis flags), execution of monitoring dialog actions, texts in config/texts/dialog.yaml Refs #10 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
50 lines
2.5 KiB
Python
50 lines
2.5 KiB
Python
"""Crisis path of the dialog (guardrail #2): fixed texts only, never generated by the LLM.
|
|
|
|
Decisions:
|
|
* The crisis script (`app.core.safety.crisis_reply`, `config/texts/crisis.yaml`) is sent verbatim, once per
|
|
open crisis. "Open" = `data["crisis"]["open"]` — set when the crisis starts, cleared only by the doctor's
|
|
acknowledgement (`doctor_ack`), so a patient repeating the phrase does not get the same long script again:
|
|
they get a short fixed "я рядом, врач уже уведомлён" + emergency resources instead.
|
|
* Entering the crisis drops unfinished scale sessions (their answered items are already stored item by item):
|
|
nothing else is asked while the patient is in crisis, except the C-SSRS screen.
|
|
* Without valid consent (before consent, after refusal/revocation, minors) we still answer a crisis message —
|
|
protecting life is the one processing ground that does not need consent (152-FZ art. 10 p. 2.3) — but with
|
|
a variant WITHOUT "your doctor has been notified", because no doctor is notified, and nothing is stored.
|
|
"""
|
|
|
|
from typing import Any
|
|
|
|
from app.core.safety import CrisisResource, CrisisScript, crisis_reply
|
|
from app.dialog.contracts import Reply
|
|
from app.dialog.fsm import Ctx, Step
|
|
|
|
|
|
def crisis_open(data: dict[str, Any]) -> bool:
|
|
return bool((data.get("crisis") or {}).get("open"))
|
|
|
|
|
|
def _resource_lines(resources: list[CrisisResource]) -> list[str]:
|
|
return [f"• {r.name}: {r.phone}" + (f" ({r.note})" if r.note else "") for r in resources]
|
|
|
|
|
|
def open_crisis(ctx: Ctx, categories: list[str], source: str) -> Step:
|
|
"""Start the crisis (full script) or, if one is already open, send the short repeat text.
|
|
|
|
`source` is a code for the dashboard/audit trail: "safety:lexicon", "safety:llm", "scale:phq9", "rule:<id>".
|
|
"""
|
|
script = crisis_reply(categories)
|
|
if crisis_open(ctx.data):
|
|
text = "\n".join([ctx.texts.crisis.repeat, *_resource_lines(script.resources)])
|
|
return Step(replies=[Reply(text=text, kind="crisis")])
|
|
crisis = {"open": True, "opened_at": ctx.now.isoformat(), "source": source, "variant": script.variant}
|
|
return Step(
|
|
replies=[Reply(text=script.text, kind="crisis")],
|
|
next_state="crisis",
|
|
data={"crisis": crisis, "scales": None},
|
|
)
|
|
|
|
|
|
def no_consent_crisis_text(script: CrisisScript) -> str:
|
|
"""Crisis script without the "doctor notified" line — for people we may not (or did not) notify about."""
|
|
return "\n".join([script.body, "", *_resource_lines(script.resources), "", script.grounding])
|