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>
75 lines
2.8 KiB
Python
75 lines
2.8 KiB
Python
"""Dialog contracts (frozen, #34). Channel-agnostic: Telegram and the web demo console render the same replies.
|
|
|
|
Lane A implements `DialogEngine` (#10); lane B calls it from the API (#12) and the scheduler (#13).
|
|
Change only via a separate small contract PR.
|
|
"""
|
|
|
|
from collections.abc import Callable
|
|
from importlib import import_module, util
|
|
from typing import Literal, Protocol
|
|
|
|
from pydantic import BaseModel, Field
|
|
|
|
SystemEventKind = Literal[
|
|
"med_reminder", # dose is due: params {"dose_id", "drug", "dose"}
|
|
"daily_checkin", # time for the daily check-in
|
|
"scale_due", # scheduled / unscheduled scale session: params {"scale", "scheduled"}
|
|
"reach_out", # silence or missed doses — caring contact
|
|
"doctor_ack", # doctor acknowledged a crisis alert: lets the FSM leave `crisis`
|
|
]
|
|
|
|
|
|
class Button(BaseModel):
|
|
"""Abstract button; channels render it (inline keyboard / web button). `data` is echoed back on press."""
|
|
|
|
label: str
|
|
data: str
|
|
|
|
|
|
class Event(BaseModel):
|
|
"""Incoming event for a patient. Exactly one of text / button / system is meaningful per `type`."""
|
|
|
|
type: Literal["text", "button", "system", "start"]
|
|
text: str | None = None # raw patient text — the engine scrubs it before anything else
|
|
button: str | None = None # Button.data of the pressed button
|
|
system: SystemEventKind | None = None
|
|
params: dict[str, object] = Field(default_factory=dict) # e.g. {"code": "..."} for type="start"
|
|
|
|
|
|
class Reply(BaseModel):
|
|
"""Outgoing message. `kind` lets channels style crisis/safety messages differently."""
|
|
|
|
text: str
|
|
buttons: list[list[Button]] = Field(default_factory=list) # rows of buttons
|
|
kind: Literal["normal", "question", "crisis", "safety_plan", "info"] = "normal"
|
|
|
|
|
|
class DialogEngine(Protocol):
|
|
async def handle(self, token: str, event: Event) -> list[Reply]: ...
|
|
|
|
|
|
class EchoEngine:
|
|
"""Placeholder until #10 lands: lets the API and the demo console be built and tested now."""
|
|
|
|
async def handle(self, token: str, event: Event) -> list[Reply]:
|
|
if event.type == "system":
|
|
return [Reply(text=f"[stub] system event: {event.system}", kind="info")]
|
|
return [Reply(text="[stub] движок диалога появится в #10", kind="info")]
|
|
|
|
|
|
_factory: Callable[[], DialogEngine] = EchoEngine
|
|
|
|
|
|
def register_engine(factory: Callable[[], DialogEngine]) -> None:
|
|
"""Lane A calls this at import time of `app.dialog.engine` to replace the echo stub."""
|
|
global _factory
|
|
_factory = factory
|
|
|
|
|
|
def get_engine() -> DialogEngine:
|
|
# Importing the engine module registers the real factory; absent module → echo stub.
|
|
# find_spec instead of try/except ImportError: real import errors inside the module must surface.
|
|
if util.find_spec("app.dialog.engine") is not None:
|
|
import_module("app.dialog.engine")
|
|
return _factory()
|