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>
74 lines
3.9 KiB
Python
74 lines
3.9 KiB
Python
"""Execution of *dialog* actions returned by `MonitoringService.after_data_change()` (ADR-007).
|
||
|
||
The rules (lane B, `config/rules.yaml`) decide WHAT happens; this module only turns an already chosen action
|
||
into replies. `alert` / `enhanced_monitoring` are handled by lane B itself (persisted + doctor notified), so they
|
||
are ignored here. While a crisis is open only crisis-relevant actions (crisis, safety plan, risk screen) reach
|
||
the patient: no medication tips or visit offers in the middle of a crisis.
|
||
"""
|
||
|
||
from collections.abc import Iterable
|
||
from typing import Any
|
||
|
||
from app.core.rules import Action
|
||
from app.db.clinical import SafetyPlan
|
||
from app.dialog.contracts import Button, Reply
|
||
from app.dialog.crisis import crisis_open, open_crisis
|
||
from app.dialog.fsm import Ctx, Step
|
||
from app.dialog.scale_session import active_scale, add_pending
|
||
|
||
# med_info topic → phrase intent of `config/texts/phrases.yaml` (the LLM only rephrases the template).
|
||
_MED_INFO_INTENTS = {"delayed_effect": "quit_intent", "alcohol_interaction": "alcohol_info", "nausea": "nausea_info"}
|
||
_CRISIS_SAFE = {"crisis", "show_safety_plan", "schedule_scale"}
|
||
|
||
|
||
def _plan_line(step: Any) -> str:
|
||
if isinstance(step, dict):
|
||
return ": ".join(str(v) for v in step.values() if v)
|
||
return str(step)
|
||
|
||
|
||
async def safety_plan_reply(ctx: Ctx, block: str | None = None) -> Reply:
|
||
"""The plan composed with the doctor (`safety_plans`), or the Stanley–Brown template from config."""
|
||
row = await ctx.session.get(SafetyPlan, ctx.token)
|
||
steps = [_plan_line(s) for s in row.steps] if row is not None and row.steps else ctx.texts.safety_plan.template
|
||
lines = [ctx.texts.safety_plan.title, *(f"{i}. {s}" for i, s in enumerate(steps, 1))]
|
||
if block == "means_restriction":
|
||
lines += ["", ctx.texts.safety_plan.means_restriction]
|
||
return Reply(text="\n".join(lines), kind="safety_plan")
|
||
|
||
|
||
async def execute(ctx: Ctx, actions: Iterable[Action]) -> Step:
|
||
step = Step()
|
||
shown_plan = False
|
||
for action in actions:
|
||
if crisis_open({**ctx.data, **step.data}) and action.type not in _CRISIS_SAFE:
|
||
continue
|
||
if action.type == "crisis":
|
||
if not crisis_open({**ctx.data, **step.data}):
|
||
step = step.then(open_crisis(ctx, categories=[], source=f"rule:{action.code}"))
|
||
elif action.type == "show_safety_plan" and not shown_plan:
|
||
shown_plan = True
|
||
step = step.then(Step(replies=[await safety_plan_reply(ctx, action.params.get("block"))]))
|
||
elif action.type == "schedule_scale" and action.params.get("scale"):
|
||
scale_id = str(action.params["scale"])
|
||
if active_scale(step.next_state or ctx.state) == scale_id:
|
||
continue # already being asked
|
||
data = {**ctx.data, **step.data}
|
||
pending = add_pending(data, scale_id, bool(action.params.get("scheduled", False)))
|
||
step = step.then(Step(data={"pending_scales": pending}))
|
||
elif action.type == "med_info":
|
||
intent = _MED_INFO_INTENTS.get(str(action.params.get("topic")), "default")
|
||
text = await ctx.gateway.phrase(intent, {})
|
||
step = step.then(Step(replies=[Reply(text=text, kind="info")]))
|
||
elif action.type == "suggest_visit":
|
||
text = await ctx.gateway.phrase("book_visit_offer", {})
|
||
button = Button(label=ctx.texts.visit.button, data="menu:visit")
|
||
step = step.then(Step(replies=[Reply(text=text, buttons=[[button]], kind="info")]))
|
||
elif action.type == "reach_out":
|
||
topic = str(action.params.get("topic", "default"))
|
||
texts = ctx.texts.reach_out
|
||
text = {"silence": texts.silence, "missed_doses": texts.missed_doses}.get(topic, texts.default)
|
||
step = step.then(Step(replies=[Reply(text=text)]))
|
||
# alert / enhanced_monitoring: lane B persists and notifies; nothing to say to the patient.
|
||
return step
|