ElebobaandClaude Opus 5.5
dd67905626
CI / ruff · pytest (pull_request) Successful in 2m22s
- An acked crisis is not raised again for the same signal: re-raised only if a newer triggering scale exists (seen live: AUDIT-C 0/0/0 re-fired the acked PHQ-9 item 9 crisis). - In an open crisis every non-acute message gets a reply (agent for questions, support otherwise) plus emergency contacts; acute messages still get the fixed script. - Questions and medicine mentions without «?» are answered mid-questionnaire; free-text steps (check-in, side effects, question states) keep parsing their input. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
192 lines
8.7 KiB
Python
192 lines
8.7 KiB
Python
"""Monitoring service (#38): patient snapshot → deterministic rules → persisted alerts (ADR-007).
|
|
|
|
`after_data_change(token)` is called by the dialog engine after it commits clinical data (and by the scheduler
|
|
on silence). It builds a `PatientSnapshot`, runs `app.core.rules.evaluate`, then in one transaction:
|
|
* writes every `alert`/`crisis` action to `alerts` (a crisis is an `acute` alert) and, for a crisis,
|
|
an audit record `crisis_acute` (guardrail #6);
|
|
* applies `enhanced_monitoring` to the profile;
|
|
* refreshes the cached `status_color` of the profile (ADR-006; the source of truth stays the open alerts).
|
|
After the commit the doctor is notified through `integrations.notifier` (a crisis also goes to the doctor on
|
|
duty), and only the *dialog* actions are returned to the engine — it never creates alerts itself.
|
|
|
|
A crisis rule whose acute alert is still open stays silent: the crisis script is already running and the
|
|
doctor already knows, so it is neither restarted nor re-alerted (`rules.evaluate` never suppresses crises).
|
|
|
|
The virtual clock offset is persisted in `clock_state` (ADR-005) by `init_clock_persistence()`.
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
from datetime import datetime, timedelta
|
|
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core import clock
|
|
from app.core.rules import Action, evaluate, patient_status
|
|
from app.core.types import AlertLevel
|
|
from app.db import repo
|
|
from app.db.audit import audit
|
|
from app.db.session import get_clinical_session
|
|
from app.integrations.base import integrations
|
|
from app.services.contracts import register_monitoring
|
|
from app.services.followup import schedule_post_crisis_call
|
|
from app.services.snapshot import build_snapshot
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
SYSTEM_ACTOR = "system"
|
|
# Notification address of the doctor on duty: acute alerts must reach someone even when the attending doctor
|
|
# is off (SPEC: crisis → immediate escalation). The notifier maps it to the duty channel.
|
|
ON_DUTY_DOCTOR = "on-duty"
|
|
# Persisted by the service itself, never passed to the dialog engine.
|
|
_SERVICE_ACTIONS = frozenset({"alert", "enhanced_monitoring"})
|
|
|
|
|
|
async def refresh_status(session: AsyncSession, token: str) -> str:
|
|
"""Recompute and cache the dashboard colour from open alerts (also call it after an ack)."""
|
|
alerts = await repo.list_alerts(session, token=token, open_only=True)
|
|
status = patient_status(AlertLevel[a.level.upper()] for a in alerts)
|
|
await repo.upsert_profile(session, token=token, status_color=status.color)
|
|
return status.color
|
|
|
|
|
|
# Which data a crisis rule is triggered by: a crisis for the same code is raised again only when such data is
|
|
# newer than the last alert with that code (acked or not) — otherwise an acked PHQ-9 item 9 re-fired on every
|
|
# data change for 24 h (seen live: AUDIT-C answers 0/0/0 → crisis again).
|
|
_CRISIS_TRIGGERS: dict[str, tuple[str, ...]] = {
|
|
"phq9_item9": ("phq9",),
|
|
"cssrs_high": ("cssrs_screen",),
|
|
"cssrs_moderate": ("cssrs_screen",),
|
|
"weapon_access_risk": ("phq9", "cssrs_screen"),
|
|
}
|
|
|
|
|
|
async def _drop_handled_crises(s: AsyncSession, token: str, actions: list[Action]) -> list[Action]:
|
|
crisis_codes = {a.code for a in actions if a.type == "crisis"}
|
|
if not crisis_codes:
|
|
return actions
|
|
alerts = await repo.list_alerts(s, token=token)
|
|
results = await repo.list_scale_results(s, token)
|
|
handled: set[str] = set()
|
|
for code in crisis_codes:
|
|
last_alert = max((a.ts for a in alerts if a.code == code), default=None)
|
|
if last_alert is None:
|
|
continue
|
|
scales = _CRISIS_TRIGGERS.get(code)
|
|
last_data = max((r.ts for r in results if scales is None or r.scale in scales), default=None)
|
|
if last_data is None or last_data <= last_alert:
|
|
handled.add(code)
|
|
if handled:
|
|
log.info("monitoring token=%s crisis already handled=%s", token, sorted(handled))
|
|
return [a for a in actions if a.code not in handled]
|
|
|
|
|
|
def _drop_running_crises(actions: list[Action], open_acute: set[str]) -> list[Action]:
|
|
"""Silence every action of a crisis rule whose acute alert is still open (see module docstring)."""
|
|
running = {a.code for a in actions if a.type == "crisis" and a.code in open_acute}
|
|
return [a for a in actions if a.code not in running]
|
|
|
|
|
|
class RuleMonitoring:
|
|
async def after_data_change(self, token: str) -> list[Action]:
|
|
now = clock.now()
|
|
notify: list[tuple[str, str, str]] = [] # (doctor_id, level, text) — sent after the commit
|
|
async with get_clinical_session() as s:
|
|
profile = await repo.get_profile(s, token)
|
|
snapshot = await build_snapshot(s, token, now, profile=profile)
|
|
if profile is None or snapshot is None:
|
|
return []
|
|
open_acute = {a.code for a in await repo.list_alerts(s, token=token, open_only=True) if a.level == "acute"}
|
|
actions = _drop_running_crises(evaluate(snapshot), open_acute)
|
|
actions = await _drop_handled_crises(s, token, actions)
|
|
for action in actions:
|
|
if action.type == "enhanced_monitoring":
|
|
profile.enhanced_monitoring = True
|
|
if not action.creates_alert or action.level is None:
|
|
continue
|
|
alert = await repo.add_alert(
|
|
s,
|
|
token=token,
|
|
level=action.level,
|
|
code=action.code,
|
|
reason=action.reason_ru,
|
|
ts=now,
|
|
params=action.params,
|
|
)
|
|
level = action.level.name.lower()
|
|
notify.append((profile.doctor_id, level, action.reason_ru))
|
|
if action.type == "crisis":
|
|
await audit(
|
|
s,
|
|
actor=SYSTEM_ACTOR,
|
|
action="crisis_acute",
|
|
token=token,
|
|
details={"alert_id": alert.id, "code": action.code},
|
|
ts=now,
|
|
)
|
|
notify.append((ON_DUTY_DOCTOR, level, action.reason_ru))
|
|
await schedule_post_crisis_call(s, token, now, source=f"rule:{action.code}")
|
|
await refresh_status(s, token)
|
|
log.info("monitoring token=%s actions=%s", token, [a.code for a in actions])
|
|
for doctor_id, level, text in notify:
|
|
await _notify(doctor_id, level, text, token)
|
|
return [a for a in actions if a.type not in _SERVICE_ACTIONS]
|
|
|
|
async def is_in_crisis(self, token: str) -> bool:
|
|
async with get_clinical_session() as s:
|
|
alerts = await repo.list_alerts(s, token=token, open_only=True)
|
|
return any(a.level == "acute" for a in alerts)
|
|
|
|
|
|
async def _notify(doctor_id: str, level: str, text: str, token: str) -> None:
|
|
# The alert is already committed and visible on the dashboard; a failing channel must not undo it.
|
|
try:
|
|
await integrations.notifier.send(doctor_id, level, text, token) # type: ignore[arg-type]
|
|
except Exception as exc: # noqa: BLE001 — any channel error
|
|
log.warning("notify failed token=%s level=%s error=%s", token, level, type(exc).__name__)
|
|
|
|
|
|
register_monitoring(RuleMonitoring)
|
|
|
|
|
|
# ---------------------------------------------------------------------------------- demo clock persistence
|
|
|
|
_writes: set[asyncio.Task[None]] = set()
|
|
|
|
|
|
async def _write_offset() -> None:
|
|
# Writes the offset current at write time, so out-of-order tasks still leave the latest value stored.
|
|
try:
|
|
async with get_clinical_session() as s:
|
|
await repo.save_clock_offset(s, clock.offset().total_seconds())
|
|
except (OSError, SQLAlchemyError) as exc:
|
|
log.warning("clock offset not saved error=%s", type(exc).__name__)
|
|
|
|
|
|
def _save_offset(_: timedelta) -> None:
|
|
"""Sync hook for `clock.register_persistence`: schedules the DB write on the running loop."""
|
|
try:
|
|
loop = asyncio.get_running_loop()
|
|
except RuntimeError:
|
|
log.warning("clock offset not saved: no running event loop")
|
|
return
|
|
task = loop.create_task(_write_offset())
|
|
_writes.add(task)
|
|
task.add_done_callback(_writes.discard)
|
|
|
|
|
|
async def flush_clock_writes() -> None:
|
|
"""Wait for pending offset writes (tests, shutdown)."""
|
|
if _writes:
|
|
await asyncio.gather(*list(_writes))
|
|
|
|
|
|
async def init_clock_persistence() -> datetime:
|
|
"""Restore the virtual clock offset from `clock_state` and save every later change there."""
|
|
async with get_clinical_session() as s:
|
|
stored = timedelta(seconds=await repo.load_clock_offset(s))
|
|
clock.register_persistence(lambda: stored, _save_offset)
|
|
log.info("clock offset restored seconds=%s", int(stored.total_seconds()))
|
|
return clock.now()
|