feat(backend): monitoring service — snapshot, rules, alerts, status colour, clock persistence

- app/services/snapshot.py: PatientSnapshot from the clinical DB (field mapping, age, therapy day,
  recent C-SSRS risk, weapon access from dialog facts)
- app/services/monitoring.py: RuleMonitoring registered via register_monitoring; persists alert/crisis
  actions, crisis_acute audit, enhanced_monitoring flag, cached status_color, notifies the doctor (and
  the doctor on duty for crises) after commit; a running crisis is not restarted
- demo clock offset persisted in clock_state, restored on scheduler start
- tests: Andrey's 13 days through the service → G G G Y Y G G G G Y Y Y R

Refs #38

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

app/services/monitoring.py

@@ -0,0 +1,157 @@
+"""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.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
+
+
+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)
+            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 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()