ElebobaandClaude Opus 5.5
b4731fb1d6
CI / ruff · pytest (pull_request) Successful in 22s
Freeze interfaces between lanes A (bot), B (backend), C (dashboard) so three developers work in parallel without touching each other's files: dialog Event/Reply + engine registry with echo stub, integration protocols with in-memory defaults, MonitoringService with noop stub, /api/v1 schemas with fixture-backed routes, frozen main.py wiring, config/modules.yaml + due_scales(), DAR-5, tzdata, Andrey trajectory, lanes in WORKFLOW, ADR-010. Refs #34 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
48 lines
1.6 KiB
Python
48 lines
1.6 KiB
Python
"""Monitoring service contract (frozen, #34).
|
|
|
|
Split of responsibilities (ADR-007):
|
|
* Lane B (#38) implements `MonitoringService`: builds a patient snapshot from the DB, runs the deterministic
|
|
rules (`app.core.rules.evaluate`), persists `alert`/`crisis` actions and notifies the doctor itself.
|
|
* It returns only the *dialog* actions (schedule_scale, med_info, reach_out, show_safety_plan, suggest_visit,
|
|
crisis) so the engine (lane A, #10) can react in the conversation. The engine never creates alerts.
|
|
"""
|
|
|
|
from collections.abc import Callable
|
|
from importlib import import_module, util
|
|
from typing import Protocol
|
|
|
|
from app.core.rules import Action
|
|
|
|
|
|
class MonitoringService(Protocol):
|
|
async def after_data_change(self, token: str) -> list[Action]: ...
|
|
|
|
async def is_in_crisis(self, token: str) -> bool:
|
|
"""True while an `acute` alert is open (not acked by the doctor) — FSM stays in `crisis`."""
|
|
...
|
|
|
|
|
|
class NoopMonitoring:
|
|
"""Placeholder until #38 lands."""
|
|
|
|
async def after_data_change(self, token: str) -> list[Action]:
|
|
return []
|
|
|
|
async def is_in_crisis(self, token: str) -> bool:
|
|
return False
|
|
|
|
|
|
_factory: Callable[[], MonitoringService] = NoopMonitoring
|
|
|
|
|
|
def register_monitoring(factory: Callable[[], MonitoringService]) -> None:
|
|
global _factory
|
|
_factory = factory
|
|
|
|
|
|
def get_monitoring() -> MonitoringService:
|
|
# find_spec instead of try/except ImportError: real import errors inside the module must surface.
|
|
if util.find_spec("app.services.monitoring") is not None:
|
|
import_module("app.services.monitoring")
|
|
return _factory()
|