app/db/audit.py

@@ -0,0 +1,41 @@
+"""Audit log (guardrail #6): every doctor access to patient data and every safety-critical system action.
+
+Lives in the clinical DB (ADR-003) and references patients only by token.
+"""
+
+from datetime import datetime
+from typing import Any
+
+from sqlalchemy import String, Text
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db.clinical import TOKEN_LEN, BigId, ClinicalBase
+from app.db.types import JSONType, TZDateTime
+
+
+class AuditLog(ClinicalBase):
+    __tablename__ = "audit_log"
+
+    id: Mapped[int] = mapped_column(BigId, primary_key=True, autoincrement=True)
+    ts: Mapped[datetime] = mapped_column(TZDateTime, index=True)
+    actor: Mapped[str] = mapped_column(String(64))  # doctor_id | system
+    action: Mapped[str] = mapped_column(Text)  # view_patient | ack_alert | crisis_acute | export ...
+    token: Mapped[str | None] = mapped_column(String(TOKEN_LEN), index=True)
+    details: Mapped[dict[str, Any] | None] = mapped_column(JSONType)
+
+
+async def audit(
+    session: AsyncSession,
+    *,
+    actor: str,
+    action: str,
+    token: str | None,
+    details: dict[str, Any] | None = None,
+    ts: datetime,
+) -> AuditLog:
+    """Append an audit record in the caller's transaction. `details` must not contain PII or raw text."""
+    entry = AuditLog(ts=ts, actor=actor, action=action, token=token, details=details)
+    session.add(entry)
+    await session.flush()
+    return entry

app/db/bootstrap.py

@@ -0,0 +1,17 @@
+"""Schema bootstrap: `create_all` on both databases at startup (ADR-003: no Alembic in the MVP).
+
+The only module (besides registration code) that touches both schemas — and only their metadata.
+"""
+
+from app.db import audit as _audit  # noqa: F401  (registers audit_log in clinical metadata)
+from app.db.clinical import ClinicalBase
+from app.db.identity import IdentityBase
+from app.db.session import clinical_engine, identity_engine
+
+
+async def init_db() -> None:
+    """Create missing tables in the identity and clinical databases. Idempotent."""
+    async with identity_engine().begin() as conn:
+        await conn.run_sync(IdentityBase.metadata.create_all)
+    async with clinical_engine().begin() as conn:
+        await conn.run_sync(ClinicalBase.metadata.create_all)

app/db/clinical.py

@@ -0,0 +1,194 @@
+"""Clinical DB schema: pseudonymised health data. SPEC §5, ADR-002, ADR-003, ADR-006.
+
+Every row is keyed by `token` (32 hex chars of HMAC(PSEUDO_KEY, SNILS)), never by name/SNILS/phone/chat id.
+Free text is stored only after `scrub()` (`*_scrubbed` columns). This module must not import
+`app.db.identity` (guardrail #5, checked in tests/test_db.py); `audit_log` also lives in this DB.
+
+Business timestamps have no server defaults: callers pass time from `app.core.clock` (ADR-005), so the
+demo virtual clock is reflected consistently in data, rules and the dashboard.
+"""
+
+from datetime import date, datetime
+from typing import Any
+
+from sqlalchemy import (
+    BigInteger,
+    Boolean,
+    Date,
+    Float,
+    ForeignKey,
+    Index,
+    Integer,
+    SmallInteger,
+    String,
+    Text,
+    UniqueConstraint,
+)
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
+
+from app.db.types import JSONType, TZDateTime
+
+TOKEN_LEN = 32
+# BIGSERIAL on Postgres; SQLite autoincrements only INTEGER PRIMARY KEY.
+BigId = BigInteger().with_variant(Integer(), "sqlite")
+
+
+class ClinicalBase(DeclarativeBase):
+    """Metadata of the clinical DB (incl. audit_log); never mixed with identity metadata."""
+
+
+class PatientProfile(ClinicalBase):
+    """Clinical view of a patient: no name, SNILS or contacts — only what monitoring needs."""
+
+    __tablename__ = "patient_profiles"
+
+    token: Mapped[str] = mapped_column(String(TOKEN_LEN), primary_key=True)
+    doctor_id: Mapped[str] = mapped_column(String(64), index=True)
+    modules: Mapped[list[str]] = mapped_column(JSONType, default=list)  # veteran | depression | family
+    therapy_start: Mapped[date | None] = mapped_column(Date)
+    next_visit: Mapped[date | None] = mapped_column(Date)
+    birth_year: Mapped[int | None] = mapped_column(SmallInteger)
+    sex: Mapped[str | None] = mapped_column(String(1))
+    # Set by rules (e.g. after an acute episode): denser check-ins and lower silence thresholds.
+    enhanced_monitoring: Mapped[bool] = mapped_column(Boolean, default=False)
+    # Cache for the dashboard list; the source of truth is open alerts (colour is derived, ADR-006).
+    status_color: Mapped[str | None] = mapped_column(String(8))
+
+
+class ItemResponse(ClinicalBase):
+    """Single scale item answer (PHQ-9, C-SSRS...) from buttons, LLM extraction or clinician."""
+
+    __tablename__ = "item_responses"
+    __table_args__ = (Index("ix_item_responses_token_ts", "token", "ts"),)
+
+    id: Mapped[int] = mapped_column(BigId, primary_key=True, autoincrement=True)
+    token: Mapped[str] = mapped_column(String(TOKEN_LEN))
+    ts: Mapped[datetime] = mapped_column(TZDateTime)
+    scale: Mapped[str] = mapped_column(String(16))  # phq9 | gad7 | pcl5 | cssrs | auditc ...
+    item_id: Mapped[str] = mapped_column(String(32))
+    score: Mapped[int | None] = mapped_column(SmallInteger)
+    confidence: Mapped[float | None] = mapped_column(Float)
+    evidence: Mapped[str | None] = mapped_column(Text)  # scrubbed phrase supporting an LLM-extracted score
+    source: Mapped[str] = mapped_column(String(16))  # llm | button | clinician
+
+
+class ScaleResult(ClinicalBase):
+    __tablename__ = "scale_results"
+    __table_args__ = (Index("ix_scale_results_token_ts", "token", "ts"),)
+
+    id: Mapped[int] = mapped_column(BigId, primary_key=True, autoincrement=True)
+    token: Mapped[str] = mapped_column(String(TOKEN_LEN))
+    ts: Mapped[datetime] = mapped_column(TZDateTime)
+    scale: Mapped[str] = mapped_column(String(16))
+    period_start: Mapped[date | None] = mapped_column(Date)
+    period_end: Mapped[date | None] = mapped_column(Date)
+    total: Mapped[int] = mapped_column(SmallInteger)
+    severity: Mapped[str] = mapped_column(String(32))
+    complete: Mapped[bool] = mapped_column(Boolean)
+    # PHQ-9 item 9 (suicidal ideation) kept separately: any value >= 1 escalates regardless of the total.
+    item9: Mapped[int | None] = mapped_column(SmallInteger)
+    # False = unscheduled administration triggered by a rule (e.g. mood <= 3 for three days, ADR-002).
+    scheduled: Mapped[bool] = mapped_column(Boolean, default=True)
+    flags: Mapped[dict[str, Any]] = mapped_column(JSONType, default=dict)
+
+
+class DailyCheckin(ClinicalBase):
+    """One row per patient per (virtual) day, filled in several steps of the daily dialog (ADR-002)."""
+
+    __tablename__ = "daily_checkins"
+    __table_args__ = (UniqueConstraint("token", "day", name="uq_daily_checkins_token_day"),)
+
+    id: Mapped[int] = mapped_column(BigId, primary_key=True, autoincrement=True)
+    token: Mapped[str] = mapped_column(String(TOKEN_LEN))
+    day: Mapped[date] = mapped_column(Date)
+    mood: Mapped[int | None] = mapped_column(SmallInteger)  # 0..10
+    med_taken: Mapped[bool | None] = mapped_column(Boolean)
+    sleep_bucket: Mapped[str | None] = mapped_column(String(16))
+    alcohol_bucket: Mapped[str | None] = mapped_column(String(16))
+    side_effects: Mapped[list[str]] = mapped_column(JSONType, default=list)
+    quit_intent: Mapped[bool | None] = mapped_column(Boolean)  # wants to stop medication — doctor flag
+    nightmares: Mapped[bool | None] = mapped_column(Boolean)
+    free_text_scrubbed: Mapped[str | None] = mapped_column(Text)
+    updated_at: Mapped[datetime] = mapped_column(TZDateTime)
+
+
+class Prescription(ClinicalBase):
+    __tablename__ = "prescriptions"
+
+    id: Mapped[int] = mapped_column(BigId, primary_key=True, autoincrement=True)
+    token: Mapped[str] = mapped_column(String(TOKEN_LEN), index=True)
+    drug: Mapped[str] = mapped_column(String(128))  # INN (МНН)
+    dose: Mapped[str] = mapped_column(String(64))
+    schedule: Mapped[dict[str, Any]] = mapped_column(JSONType)  # {"times": ["08:00", "20:00"]}
+    start_date: Mapped[date | None] = mapped_column(Date)
+    end_date: Mapped[date | None] = mapped_column(Date)
+
+
+class Dose(ClinicalBase):
+    __tablename__ = "doses"
+    __table_args__ = (Index("ix_doses_token_planned", "token", "planned_at"),)
+
+    id: Mapped[int] = mapped_column(BigId, primary_key=True, autoincrement=True)
+    token: Mapped[str] = mapped_column(String(TOKEN_LEN))
+    prescription_id: Mapped[int] = mapped_column(ForeignKey("prescriptions.id"))
+    planned_at: Mapped[datetime] = mapped_column(TZDateTime)
+    status: Mapped[str] = mapped_column(String(16), default="pending")  # taken | missed | skipped | pending
+    answered_at: Mapped[datetime | None] = mapped_column(TZDateTime)
+    side_effects: Mapped[str | None] = mapped_column(Text)
+
+
+class Alert(ClinicalBase):
+    """Output of the deterministic rules engine (ADR-007). Level is stored, UI colour is derived (ADR-006)."""
+
+    __tablename__ = "alerts"
+    __table_args__ = (Index("ix_alerts_token_ts", "token", "ts"),)
+
+    id: Mapped[int] = mapped_column(BigId, primary_key=True, autoincrement=True)
+    token: Mapped[str] = mapped_column(String(TOKEN_LEN))
+    ts: Mapped[datetime] = mapped_column(TZDateTime)
+    level: Mapped[str] = mapped_column(String(16))  # acute | high | moderate | info (AlertLevel name)
+    code: Mapped[str] = mapped_column(String(64))  # rule id from config/rules.yaml
+    reason: Mapped[str] = mapped_column(Text)
+    params: Mapped[dict[str, Any]] = mapped_column(JSONType, default=dict)
+    acked_by: Mapped[str | None] = mapped_column(String(64))
+    acked_at: Mapped[datetime | None] = mapped_column(TZDateTime)
+
+
+class Message(ClinicalBase):
+    __tablename__ = "messages"
+    __table_args__ = (Index("ix_messages_token_ts", "token", "ts"),)
+
+    id: Mapped[int] = mapped_column(BigId, primary_key=True, autoincrement=True)
+    token: Mapped[str] = mapped_column(String(TOKEN_LEN))
+    ts: Mapped[datetime] = mapped_column(TZDateTime)
+    direction: Mapped[str] = mapped_column(String(3))  # in | out
+    text_scrubbed: Mapped[str] = mapped_column(Text)
+    fsm_state: Mapped[str | None] = mapped_column(String(64))
+
+
+class DialogState(ClinicalBase):
+    """FSM state of the dialog engine, persisted instead of Redis (ADR-003)."""
+
+    __tablename__ = "dialog_state"
+
+    token: Mapped[str] = mapped_column(String(TOKEN_LEN), primary_key=True)
+    state: Mapped[str] = mapped_column(String(64))
+    data: Mapped[dict[str, Any]] = mapped_column(JSONType, default=dict)
+    updated_at: Mapped[datetime] = mapped_column(TZDateTime)
+
+
+class SafetyPlan(ClinicalBase):
+    __tablename__ = "safety_plans"
+
+    token: Mapped[str] = mapped_column(String(TOKEN_LEN), primary_key=True)
+    steps: Mapped[list[Any]] = mapped_column(JSONType, default=list)
+    updated_at: Mapped[datetime] = mapped_column(TZDateTime)
+
+
+class ClockState(ClinicalBase):
+    """Single row (id=1): virtual clock offset so the demo timeline survives restarts (ADR-005)."""
+
+    __tablename__ = "clock_state"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    offset_seconds: Mapped[float] = mapped_column(Float, default=0.0)

app/db/identity.py

@@ -0,0 +1,63 @@
+"""Identity DB schema: direct identifiers (152-FZ personal data). SPEC §5, ADR-003.
+
+Why a separate database: a leak or over-broad query on the clinical side must not reveal who the patient
+is. Names, SNILS, phones and Telegram chat ids live only here; the clinical DB stores health data keyed by
+`token` = HMAC(PSEUDO_KEY, SNILS) (`app.core.privacy.pseudo_id`). Without the key and this DB the clinical
+data is pseudonymised. In compose each DB has its own Postgres role with no access to the other one.
+
+Only registration and doctor-notification code may import this module (guardrail #5).
+"""
+
+from datetime import datetime
+
+from sqlalchemy import BigInteger, Boolean, ForeignKey, SmallInteger, String, Text
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
+
+from app.db.types import TZDateTime
+
+TOKEN_LEN = 32
+
+
+class IdentityBase(DeclarativeBase):
+    """Metadata of the identity DB only; never mixed with clinical metadata."""
+
+
+class Patient(IdentityBase):
+    __tablename__ = "patients"
+
+    token: Mapped[str] = mapped_column(String(TOKEN_LEN), primary_key=True)
+    display_name: Mapped[str] = mapped_column(Text)
+    snils: Mapped[str] = mapped_column(String(14), unique=True)
+    phone: Mapped[str | None] = mapped_column(String(32))
+    tg_chat_id: Mapped[int | None] = mapped_column(BigInteger, unique=True)
+    birth_year: Mapped[int | None] = mapped_column(SmallInteger)
+    sex: Mapped[str | None] = mapped_column(String(1))  # "m" | "f"
+    clinic_id: Mapped[str] = mapped_column(String(64))
+    doctor_id: Mapped[str] = mapped_column(String(64), index=True)
+    # NULL consent_at = no consent: processing of health data is forbidden (152-FZ art. 10).
+    consent_at: Mapped[datetime | None] = mapped_column(TZDateTime)
+    consent_ver: Mapped[str | None] = mapped_column(String(32))
+    revoked_at: Mapped[datetime | None] = mapped_column(TZDateTime)
+    created_at: Mapped[datetime] = mapped_column(TZDateTime)
+
+
+class Doctor(IdentityBase):
+    __tablename__ = "doctors"
+
+    doctor_id: Mapped[str] = mapped_column(String(64), primary_key=True)
+    full_name: Mapped[str] = mapped_column(Text)
+    clinic_id: Mapped[str] = mapped_column(String(64))
+    notify_chat: Mapped[int | None] = mapped_column(BigInteger)
+    on_call: Mapped[bool] = mapped_column(Boolean, default=False)
+
+
+class ConnectCode(IdentityBase):
+    """One-time code the doctor gives the patient (QR) to bind the bot chat to the patient record."""
+
+    __tablename__ = "connect_codes"
+
+    code: Mapped[str] = mapped_column(String(32), primary_key=True)
+    token: Mapped[str] = mapped_column(ForeignKey("patients.token"), index=True)
+    doctor_id: Mapped[str] = mapped_column(String(64))
+    created_at: Mapped[datetime] = mapped_column(TZDateTime)
+    used_at: Mapped[datetime | None] = mapped_column(TZDateTime)

app/db/identity_repo.py

@@ -0,0 +1,136 @@
+"""Identity-DB helpers: registration, consent, connect codes, doctor lookup.
+
+Only registration / notification code may use this module. Functions flush but never commit and take time
+as an explicit argument (ADR-005).
+"""
+
+from collections.abc import Sequence
+from datetime import datetime
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.db.identity import ConnectCode, Doctor, Patient
+
+
+async def create_patient(
+    session: AsyncSession,
+    *,
+    token: str,
+    display_name: str,
+    snils: str,
+    clinic_id: str,
+    doctor_id: str,
+    now: datetime,
+    phone: str | None = None,
+    birth_year: int | None = None,
+    sex: str | None = None,
+) -> Patient:
+    patient = Patient(
+        token=token,
+        display_name=display_name,
+        snils=snils,
+        clinic_id=clinic_id,
+        doctor_id=doctor_id,
+        created_at=now,
+        phone=phone,
+        birth_year=birth_year,
+        sex=sex,
+    )
+    session.add(patient)
+    await session.flush()
+    return patient
+
+
+async def get_patient(session: AsyncSession, token: str) -> Patient | None:
+    return await session.get(Patient, token)
+
+
+async def get_patient_by_snils(session: AsyncSession, snils: str) -> Patient | None:
+    return await session.scalar(select(Patient).where(Patient.snils == snils))
+
+
+async def get_patient_by_chat(session: AsyncSession, tg_chat_id: int) -> Patient | None:
+    return await session.scalar(select(Patient).where(Patient.tg_chat_id == tg_chat_id))
+
+
+async def bind_chat(session: AsyncSession, *, token: str, tg_chat_id: int) -> Patient | None:
+    patient = await session.get(Patient, token)
+    if patient is not None:
+        patient.tg_chat_id = tg_chat_id
+        await session.flush()
+    return patient
+
+
+async def record_consent(session: AsyncSession, *, token: str, version: str, now: datetime) -> Patient | None:
+    patient = await session.get(Patient, token)
+    if patient is not None:
+        patient.consent_at = now
+        patient.consent_ver = version
+        patient.revoked_at = None
+        await session.flush()
+    return patient
+
+
+async def revoke_consent(session: AsyncSession, *, token: str, now: datetime) -> Patient | None:
+    patient = await session.get(Patient, token)
+    if patient is not None:
+        patient.revoked_at = now
+        await session.flush()
+    return patient
+
+
+def has_active_consent(patient: Patient) -> bool:
+    """Health data may be processed only with consent given and not revoked (152-FZ art. 10)."""
+    return patient.consent_at is not None and patient.revoked_at is None
+
+
+async def upsert_doctor(
+    session: AsyncSession,
+    *,
+    doctor_id: str,
+    full_name: str,
+    clinic_id: str,
+    notify_chat: int | None = None,
+    on_call: bool = False,
+) -> Doctor:
+    doctor = await session.get(Doctor, doctor_id)
+    if doctor is None:
+        doctor = Doctor(doctor_id=doctor_id)
+        session.add(doctor)
+    doctor.full_name = full_name
+    doctor.clinic_id = clinic_id
+    doctor.notify_chat = notify_chat
+    doctor.on_call = on_call
+    await session.flush()
+    return doctor
+
+
+async def get_doctor(session: AsyncSession, doctor_id: str) -> Doctor | None:
+    return await session.get(Doctor, doctor_id)
+
+
+async def doctors_on_call(session: AsyncSession, clinic_id: str | None = None) -> Sequence[Doctor]:
+    stmt = select(Doctor).where(Doctor.on_call.is_(True))
+    if clinic_id is not None:
+        stmt = stmt.where(Doctor.clinic_id == clinic_id)
+    return (await session.scalars(stmt.order_by(Doctor.doctor_id))).all()
+
+
+async def create_connect_code(
+    session: AsyncSession, *, code: str, token: str, doctor_id: str, now: datetime
+) -> ConnectCode:
+    row = ConnectCode(code=code, token=token, doctor_id=doctor_id, created_at=now)
+    session.add(row)
+    await session.flush()
+    return row
+
+
+async def redeem_connect_code(session: AsyncSession, *, code: str, now: datetime) -> str | None:
+    """Mark a one-time connect code as used and return the patient token; None if unknown or already used."""
+    row = await session.get(ConnectCode, code)
+    if row is None or row.used_at is not None:
+        return None
+    row.used_at = now
+    await session.flush()
+    return row.token

app/db/repo.py

@@ -0,0 +1,298 @@
+"""Thin clinical-DB repository helpers. Must not import `app.db.identity` (guardrail #5).
+
+Functions work inside the caller's session (`get_clinical_session()`), flush but never commit, and take
+time as an explicit argument (ADR-005).
+"""
+
+from collections.abc import Sequence
+from datetime import date, datetime
+from typing import Any
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.types import AlertLevel, ScrubbedText
+from app.db.audit import audit
+from app.db.clinical import (
+    Alert,
+    ClockState,
+    DailyCheckin,
+    DialogState,
+    ItemResponse,
+    Message,
+    PatientProfile,
+    ScaleResult,
+)
+
+_CHECKIN_FIELDS = frozenset(
+    {
+        "mood",
+        "med_taken",
+        "sleep_bucket",
+        "alcohol_bucket",
+        "side_effects",
+        "quit_intent",
+        "nightmares",
+        "free_text_scrubbed",
+    }
+)
+_PROFILE_FIELDS = frozenset(
+    {
+        "doctor_id",
+        "modules",
+        "therapy_start",
+        "next_visit",
+        "birth_year",
+        "sex",
+        "enhanced_monitoring",
+        "status_color",
+    }
+)
+
+
+# --- patient profiles -------------------------------------------------------------------------------
+
+
+async def get_profile(session: AsyncSession, token: str) -> PatientProfile | None:
+    return await session.get(PatientProfile, token)
+
+
+async def upsert_profile(session: AsyncSession, *, token: str, **fields: Any) -> PatientProfile:
+    """Create or update a profile; only the passed fields are changed."""
+    unknown = set(fields) - _PROFILE_FIELDS
+    if unknown:
+        raise ValueError(f"unknown profile fields: {sorted(unknown)}")
+    profile = await session.get(PatientProfile, token)
+    if profile is None:
+        profile = PatientProfile(token=token, modules=[], enhanced_monitoring=False)
+        session.add(profile)
+    for key, value in fields.items():
+        setattr(profile, key, value)
+    await session.flush()
+    return profile
+
+
+async def patients_of_doctor(session: AsyncSession, doctor_id: str) -> Sequence[PatientProfile]:
+    rows = await session.scalars(
+        select(PatientProfile).where(PatientProfile.doctor_id == doctor_id).order_by(PatientProfile.token)
+    )
+    return rows.all()
+
+
+# --- daily check-ins and scales ---------------------------------------------------------------------
+
+
+async def upsert_checkin(
+    session: AsyncSession, *, token: str, day: date, now: datetime, **fields: Any
+) -> DailyCheckin:
+    """Merge answers into the (token, day) check-in.
+
+    The daily dialog fills the row step by step (mood, then medication, then the module question), so
+    only fields passed and not None overwrite stored values.
+    """
+    unknown = set(fields) - _CHECKIN_FIELDS
+    if unknown:
+        raise ValueError(f"unknown check-in fields: {sorted(unknown)}")
+    row = await session.scalar(
+        select(DailyCheckin).where(DailyCheckin.token == token, DailyCheckin.day == day)
+    )
+    if row is None:
+        row = DailyCheckin(token=token, day=day, side_effects=[], updated_at=now)
+        session.add(row)
+    for key, value in fields.items():
+        if value is not None:
+            setattr(row, key, value)
+    row.updated_at = now
+    await session.flush()
+    return row
+
+
+async def list_checkins(
+    session: AsyncSession, token: str, since: date | None = None
+) -> Sequence[DailyCheckin]:
+    stmt = select(DailyCheckin).where(DailyCheckin.token == token)
+    if since is not None:
+        stmt = stmt.where(DailyCheckin.day >= since)
+    return (await session.scalars(stmt.order_by(DailyCheckin.day))).all()
+
+
+async def add_item_response(
+    session: AsyncSession,
+    *,
+    token: str,
+    scale: str,
+    item_id: str,
+    score: int | None,
+    source: str,
+    ts: datetime,
+    confidence: float | None = None,
+    evidence: ScrubbedText | None = None,
+) -> ItemResponse:
+    row = ItemResponse(
+        token=token,
+        scale=scale,
+        item_id=item_id,
+        score=score,
+        source=source,
+        ts=ts,
+        confidence=confidence,
+        evidence=evidence,
+    )
+    session.add(row)
+    await session.flush()
+    return row
+
+
+async def add_scale_result(
+    session: AsyncSession,
+    *,
+    token: str,
+    scale: str,
+    total: int,
+    severity: str,
+    complete: bool,
+    ts: datetime,
+    item9: int | None = None,
+    scheduled: bool = True,
+    flags: dict[str, Any] | None = None,
+    period_start: date | None = None,
+    period_end: date | None = None,
+) -> ScaleResult:
+    row = ScaleResult(
+        token=token,
+        scale=scale,
+        total=total,
+        severity=severity,
+        complete=complete,
+        ts=ts,
+        item9=item9,
+        scheduled=scheduled,
+        flags=flags or {},
+        period_start=period_start,
+        period_end=period_end,
+    )
+    session.add(row)
+    await session.flush()
+    return row
+
+
+async def list_scale_results(
+    session: AsyncSession, token: str, scale: str | None = None
+) -> Sequence[ScaleResult]:
+    stmt = select(ScaleResult).where(ScaleResult.token == token)
+    if scale is not None:
+        stmt = stmt.where(ScaleResult.scale == scale)
+    return (await session.scalars(stmt.order_by(ScaleResult.ts))).all()
+
+
+# --- alerts -----------------------------------------------------------------------------------------
+
+
+async def add_alert(
+    session: AsyncSession,
+    *,
+    token: str,
+    level: AlertLevel,
+    code: str,
+    reason: str,
+    ts: datetime,
+    params: dict[str, Any] | None = None,
+) -> Alert:
+    row = Alert(token=token, level=level.name.lower(), code=code, reason=reason, ts=ts, params=params or {})
+    session.add(row)
+    await session.flush()
+    return row
+
+
+async def open_alert_codes(session: AsyncSession, token: str) -> set[str]:
+    """Codes of unacknowledged alerts — lets rules avoid raising the same alert twice."""
+    rows = await session.scalars(select(Alert.code).where(Alert.token == token, Alert.acked_at.is_(None)))
+    return set(rows.all())
+
+
+async def list_alerts(
+    session: AsyncSession, *, token: str | None = None, open_only: bool = False
+) -> Sequence[Alert]:
+    stmt = select(Alert)
+    if token is not None:
+        stmt = stmt.where(Alert.token == token)
+    if open_only:
+        stmt = stmt.where(Alert.acked_at.is_(None))
+    return (await session.scalars(stmt.order_by(Alert.ts.desc(), Alert.id.desc()))).all()
+
+
+async def ack_alert(session: AsyncSession, *, alert_id: int, doctor_id: str, ts: datetime) -> Alert | None:
+    """Doctor acknowledges an alert; the access is written to audit_log in the same transaction.
+
+    Re-acking is a no-op (first acknowledgement wins, no duplicate audit record). Unknown id -> None.
+    """
+    alert = await session.get(Alert, alert_id)
+    if alert is None or alert.acked_at is not None:
+        return alert
+    alert.acked_by = doctor_id
+    alert.acked_at = ts
+    await audit(
+        session,
+        actor=doctor_id,
+        action="ack_alert",
+        token=alert.token,
+        details={"alert_id": alert.id, "code": alert.code, "level": alert.level},
+        ts=ts,
+    )
+    return alert
+
+
+# --- dialog -----------------------------------------------------------------------------------------
+
+
+async def get_dialog_state(session: AsyncSession, token: str) -> DialogState | None:
+    return await session.get(DialogState, token)
+
+
+async def set_dialog_state(
+    session: AsyncSession, *, token: str, state: str, data: dict[str, Any], now: datetime
+) -> DialogState:
+    row = await session.get(DialogState, token)
+    if row is None:
+        row = DialogState(token=token, state=state, data=data, updated_at=now)
+        session.add(row)
+    else:
+        row.state = state
+        row.data = dict(data)  # new object so SQLAlchemy sees the JSON change
+        row.updated_at = now
+    await session.flush()
+    return row
+
+
+async def add_message(
+    session: AsyncSession,
+    *,
+    token: str,
+    direction: str,
+    text: ScrubbedText,
+    ts: datetime,
+    fsm_state: str | None = None,
+) -> Message:
+    if direction not in ("in", "out"):
+        raise ValueError("direction must be 'in' or 'out'")
+    row = Message(token=token, direction=direction, text_scrubbed=text, ts=ts, fsm_state=fsm_state)
+    session.add(row)
+    await session.flush()
+    return row
+
+
+# --- demo clock persistence (ADR-005) ----------------------------------------------------------------
+
+
+async def load_clock_offset(session: AsyncSession) -> float:
+    row = await session.get(ClockState, 1)
+    return row.offset_seconds if row else 0.0
+
+
+async def save_clock_offset(session: AsyncSession, offset_seconds: float) -> None:
+    row = await session.get(ClockState, 1)
+    if row is None:
+        session.add(ClockState(id=1, offset_seconds=offset_seconds))
+    else:
+        row.offset_seconds = offset_seconds
+    await session.flush()

app/db/session.py

@@ -0,0 +1,89 @@
+"""Engines and sessions for the two databases (ADR-003).
+
+This module knows nothing about models: clinical code can import it without pulling in the identity
+schema (guardrail #5). Schema creation lives in `app.db.bootstrap`.
+"""
+
+from collections.abc import AsyncIterator
+from contextlib import AbstractAsyncContextManager, asynccontextmanager
+
+from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine
+from sqlalchemy.pool import StaticPool
+
+from app.config import get_settings
+
+_engines: dict[str, AsyncEngine] = {}
+_makers: dict[str, async_sessionmaker[AsyncSession]] = {}
+_urls: dict[str, str] = {}
+
+
+def _make_engine(url: str) -> AsyncEngine:
+    kwargs: dict[str, object] = {}
+    if url.startswith("postgresql"):
+        # Fail fast when the DB is unreachable instead of hanging on asyncpg's 60 s default.
+        kwargs = {"pool_pre_ping": True, "connect_args": {"timeout": 5}}
+    elif url.startswith("sqlite") and (":memory:" in url or url.rstrip("/").endswith(":")):
+        # In-memory SQLite (tests): one shared connection, otherwise every connection sees an empty DB.
+        kwargs = {"poolclass": StaticPool}
+    return create_async_engine(url, **kwargs)
+
+
+def _url(name: str) -> str:
+    if name in _urls:
+        return _urls[name]
+    settings = get_settings()
+    return settings.identity_db_url if name == "identity" else settings.clinical_db_url
+
+
+def _engine(name: str) -> AsyncEngine:
+    if name not in _engines:
+        _engines[name] = _make_engine(_url(name))
+        _makers[name] = async_sessionmaker(_engines[name], expire_on_commit=False)
+    return _engines[name]
+
+
+def identity_engine() -> AsyncEngine:
+    return _engine("identity")
+
+
+def clinical_engine() -> AsyncEngine:
+    return _engine("clinical")
+
+
+async def configure(*, identity_url: str | None = None, clinical_url: str | None = None) -> None:
+    """Override DB URLs (tests, scripts). Disposes engines created so far; new ones are created lazily."""
+    await dispose()
+    _urls.clear()
+    if identity_url:
+        _urls["identity"] = identity_url
+    if clinical_url:
+        _urls["clinical"] = clinical_url
+
+
+@asynccontextmanager
+async def _session(name: str) -> AsyncIterator[AsyncSession]:
+    _engine(name)
+    async with _makers[name]() as session:
+        try:
+            yield session
+            await session.commit()
+        except BaseException:
+            await session.rollback()
+            raise
+
+
+def get_identity_session() -> AbstractAsyncContextManager[AsyncSession]:
+    """Unit of work on the identity DB: commits on success, rolls back on error."""
+    return _session("identity")
+
+
+def get_clinical_session() -> AbstractAsyncContextManager[AsyncSession]:
+    """Unit of work on the clinical DB (incl. audit_log): commits on success, rolls back on error."""
+    return _session("clinical")
+
+
+async def dispose() -> None:
+    for engine in _engines.values():
+        await engine.dispose()
+    _engines.clear()
+    _makers.clear()

app/db/types.py

@@ -0,0 +1,37 @@
+"""Portable column types: the same models run on Postgres (prod/compose) and SQLite (tests)."""
+
+from datetime import UTC, datetime
+from typing import Any
+
+from sqlalchemy import JSON, DateTime, Dialect
+from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.types import TypeDecorator
+
+# JSONB on Postgres (indexable, compact), plain JSON elsewhere. Lists are stored as JSON instead of ARRAY
+# so the schema stays portable to SQLite.
+JSONType = JSON().with_variant(JSONB(), "postgresql")
+
+
+class TZDateTime(TypeDecorator[datetime]):
+    """Timezone-aware timestamp, stored in UTC.
+
+    Rejects naive datetimes: every business timestamp must come from `app.core.clock` (ADR-005), which
+    returns aware values. SQLite drops tzinfo on storage, so values read back are re-tagged as UTC.
+    """
+
+    impl = DateTime(timezone=True)
+    cache_ok = True
+
+    def process_bind_param(self, value: datetime | None, dialect: Dialect) -> datetime | None:
+        if value is None:
+            return None
+        if value.tzinfo is None:
+            raise ValueError("naive datetime: take time from app.core.clock (ADR-005)")
+        return value.astimezone(UTC)
+
+    def process_result_value(self, value: Any, dialect: Dialect) -> datetime | None:
+        if value is None:
+            return None
+        if value.tzinfo is None:
+            return value.replace(tzinfo=UTC)
+        return value

app/main.py

@@ -5,8 +5,11 @@ from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
 
 from fastapi import FastAPI
+from sqlalchemy.exc import SQLAlchemyError
 
 from app.config import get_settings
+from app.db.bootstrap import init_db
+from app.db.session import dispose
 
 # Logs carry only tokens, states and event codes — never PII or raw message text.
 logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s")
@@ -17,8 +20,16 @@ log = logging.getLogger("app")
 async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
     settings = get_settings()
     log.info("startup llm_provider=%s demo_clock=%s", settings.llm_provider, settings.demo_clock)
-    # DB init (#6), scheduler (#13) and Telegram bot (#11) hook in here.
+    try:
+        await init_db()
+        log.info("db schema ready")
+    except (OSError, SQLAlchemyError) as exc:
+        # In compose Postgres is required and healthy before the app starts; this branch is for running the
+        # API (and its tests) without a database. Endpoints touching the DB will fail until it is reachable.
+        log.error("db init failed, continuing without db: %s", type(exc).__name__)
+    # Scheduler (#13) and Telegram bot (#11) hook in here.
     yield
+    await dispose()
 
 
 app = FastAPI(title="PTSRaiAgent API", version="0.1.0", lifespan=lifespan)

tests/test_db.py

@@ -0,0 +1,287 @@
+"""Data model tests on two separate in-memory SQLite databases (identity / clinical).
+
+Set TEST_PG_IDENTITY_URL / TEST_PG_CLINICAL_URL to run the same suite against compose Postgres.
+"""
+
+import ast
+import os
+import subprocess
+import sys
+from collections.abc import AsyncIterator, Callable
+from datetime import UTC, date, datetime, timedelta
+from pathlib import Path
+
+import pytest
+from sqlalchemy import inspect, select
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncEngine
+
+from app.core.types import AlertLevel, ScrubbedText
+from app.db import identity_repo, repo, session
+from app.db.audit import AuditLog, audit
+from app.db.bootstrap import init_db
+from app.db.clinical import ClinicalBase, DailyCheckin, Dose, Prescription
+from app.db.identity import IdentityBase
+
+DB_DIR = Path(__file__).resolve().parent.parent / "app" / "db"
+T0 = datetime(2026, 10, 3, 9, 0, tzinfo=UTC)
+TOKEN = "a" * 32
+TOKEN2 = "b" * 32
+
+
+@pytest.fixture
+async def dbs() -> AsyncIterator[None]:
+    await session.configure(
+        identity_url=os.environ.get("TEST_PG_IDENTITY_URL", "sqlite+aiosqlite:///:memory:"),
+        clinical_url=os.environ.get("TEST_PG_CLINICAL_URL", "sqlite+aiosqlite:///:memory:"),
+    )
+    if "TEST_PG_IDENTITY_URL" in os.environ:
+        async with session.identity_engine().begin() as conn:
+            await conn.run_sync(IdentityBase.metadata.drop_all)
+        async with session.clinical_engine().begin() as conn:
+            await conn.run_sync(ClinicalBase.metadata.drop_all)
+    await init_db()
+    yield
+    await session.configure()
+
+
+async def _tables(engine_fn: Callable[[], AsyncEngine]) -> set[str]:
+    async with engine_fn().connect() as conn:
+        return set(await conn.run_sync(lambda c: inspect(c).get_table_names()))
+
+
+async def test_create_all_puts_tables_in_the_right_db(dbs: None) -> None:
+    identity_tables = await _tables(session.identity_engine)
+    clinical_tables = await _tables(session.clinical_engine)
+    assert identity_tables == {"patients", "doctors", "connect_codes"}
+    assert {
+        "patient_profiles",
+        "item_responses",
+        "scale_results",
+        "daily_checkins",
+        "prescriptions",
+        "doses",
+        "alerts",
+        "messages",
+        "dialog_state",
+        "safety_plans",
+        "clock_state",
+        "audit_log",
+    } == clinical_tables
+    await init_db()  # idempotent
+
+
+async def test_identity_roundtrip_and_connect_code(dbs: None) -> None:
+    async with session.get_identity_session() as s:
+        await identity_repo.upsert_doctor(
+            s, doctor_id="d1", full_name="Доктор Тест", clinic_id="c1", on_call=True
+        )
+        await identity_repo.create_patient(
+            s,
+            token=TOKEN,
+            display_name="Пациент",
+            snils="000-000-000 00",
+            clinic_id="c1",
+            doctor_id="d1",
+            now=T0,
+        )
+        await identity_repo.create_connect_code(s, code="ABC123", token=TOKEN, doctor_id="d1", now=T0)
+    async with session.get_identity_session() as s:
+        assert await identity_repo.redeem_connect_code(s, code="ABC123", now=T0) == TOKEN
+        assert await identity_repo.redeem_connect_code(s, code="ABC123", now=T0) is None
+        assert await identity_repo.redeem_connect_code(s, code="NOPE", now=T0) is None
+        await identity_repo.bind_chat(s, token=TOKEN, tg_chat_id=42)
+        patient = await identity_repo.record_consent(s, token=TOKEN, version="v1", now=T0)
+        assert patient is not None and identity_repo.has_active_consent(patient)
+    async with session.get_identity_session() as s:
+        patient = await identity_repo.get_patient_by_chat(s, 42)
+        assert patient is not None and patient.token == TOKEN
+        assert patient.created_at == T0 and patient.created_at.tzinfo is not None
+        await identity_repo.revoke_consent(s, token=TOKEN, now=T0 + timedelta(days=1))
+        assert not identity_repo.has_active_consent(patient)
+        assert [d.doctor_id for d in await identity_repo.doctors_on_call(s)] == ["d1"]
+
+
+async def test_snils_and_chat_unique(dbs: None) -> None:
+    async with session.get_identity_session() as s:
+        await identity_repo.create_patient(
+            s, token=TOKEN, display_name="A", snils="111", clinic_id="c", doctor_id="d", now=T0
+        )
+    with pytest.raises(IntegrityError):
+        async with session.get_identity_session() as s:
+            await identity_repo.create_patient(
+                s, token=TOKEN2, display_name="B", snils="111", clinic_id="c", doctor_id="d", now=T0
+            )
+    async with session.get_identity_session() as s:
+        await identity_repo.create_patient(
+            s, token=TOKEN2, display_name="B", snils="222", clinic_id="c", doctor_id="d", now=T0
+        )
+        await identity_repo.bind_chat(s, token=TOKEN, tg_chat_id=7)
+    with pytest.raises(IntegrityError):
+        async with session.get_identity_session() as s:
+            await identity_repo.bind_chat(s, token=TOKEN2, tg_chat_id=7)
+
+
+async def test_checkin_upsert_merges_and_is_unique_per_day(dbs: None) -> None:
+    day = date(2026, 10, 3)
+    async with session.get_clinical_session() as s:
+        await repo.upsert_checkin(s, token=TOKEN, day=day, now=T0, mood=4)
+        await repo.upsert_checkin(s, token=TOKEN, day=day, now=T0, med_taken=True, side_effects=["тошнота"])
+        await repo.upsert_checkin(s, token=TOKEN, day=day + timedelta(days=1), now=T0, mood=3)
+    async with session.get_clinical_session() as s:
+        rows = await repo.list_checkins(s, TOKEN)
+        assert len(rows) == 2
+        assert (rows[0].mood, rows[0].med_taken, rows[0].side_effects) == (4, True, ["тошнота"])
+        assert [r.day for r in await repo.list_checkins(s, TOKEN, since=day + timedelta(days=1))] == [
+            day + timedelta(days=1)
+        ]
+        with pytest.raises(ValueError):
+            await repo.upsert_checkin(s, token=TOKEN, day=day, now=T0, bogus=1)
+    with pytest.raises(IntegrityError):
+        async with session.get_clinical_session() as s:
+            s.add(DailyCheckin(token=TOKEN, day=day, side_effects=[], updated_at=T0))
+
+
+async def test_scales_profiles_prescriptions(dbs: None) -> None:
+    async with session.get_clinical_session() as s:
+        await repo.upsert_profile(s, token=TOKEN, doctor_id="d1", modules=["veteran", "depression"])
+        await repo.upsert_profile(s, token=TOKEN2, doctor_id="d2")
+        await repo.upsert_profile(s, token=TOKEN, next_visit=date(2026, 10, 20))
+        await repo.add_item_response(
+            s, token=TOKEN, scale="phq9", item_id="9", score=1, source="button", ts=T0
+        )
+        await repo.add_scale_result(
+            s,
+            token=TOKEN,
+            scale="phq9",
+            total=16,
+            severity="moderately_severe",
+            complete=True,
+            ts=T0,
+            item9=1,
+            scheduled=False,
+            flags={"trigger": "low_mood_3d"},
+        )
+        rx = Prescription(token=TOKEN, drug="sertraline", dose="50 mg", schedule={"times": ["08:00"]})
+        s.add(rx)
+        await s.flush()
+        s.add(Dose(token=TOKEN, prescription_id=rx.id, planned_at=T0))
+    async with session.get_clinical_session() as s:
+        profiles = await repo.patients_of_doctor(s, "d1")
+        assert [(p.token, p.modules, p.next_visit) for p in profiles] == [
+            (TOKEN, ["veteran", "depression"], date(2026, 10, 20))
+        ]
+        (result,) = await repo.list_scale_results(s, TOKEN, "phq9")
+        assert (result.item9, result.scheduled, result.flags) == (1, False, {"trigger": "low_mood_3d"})
+        dose = await s.scalar(select(Dose))
+        assert dose is not None and dose.status == "pending" and dose.planned_at == T0
+
+
+async def test_ack_alert_writes_audit(dbs: None) -> None:
+    async with session.get_clinical_session() as s:
+        alert = await repo.add_alert(
+            s,
+            token=TOKEN,
+            level=AlertLevel.ACUTE,
+            code="phq9_item9",
+            reason="п.9 ≥ 1",
+            ts=T0,
+            params={"item9": 1},
+        )
+        await repo.add_alert(s, token=TOKEN, level=AlertLevel.INFO, code="silence", reason="молчание", ts=T0)
+        alert_id = alert.id
+        assert await repo.open_alert_codes(s, TOKEN) == {"phq9_item9", "silence"}
+    later = T0 + timedelta(minutes=5)
+    async with session.get_clinical_session() as s:
+        acked = await repo.ack_alert(s, alert_id=alert_id, doctor_id="d1", ts=later)
+        assert acked is not None and acked.acked_by == "d1"
+        await repo.ack_alert(s, alert_id=alert_id, doctor_id="d2", ts=later)  # no-op
+        assert await repo.ack_alert(s, alert_id=9999, doctor_id="d1", ts=later) is None
+    async with session.get_clinical_session() as s:
+        assert await repo.open_alert_codes(s, TOKEN) == {"silence"}
+        (stored,) = [a for a in await repo.list_alerts(s, token=TOKEN) if a.id == alert_id]
+        assert (stored.level, stored.acked_by, stored.acked_at) == ("acute", "d1", later)
+        logs = (await s.scalars(select(AuditLog))).all()
+        assert [(a.actor, a.action, a.token, a.ts) for a in logs] == [("d1", "ack_alert", TOKEN, later)]
+        assert logs[0].details == {"alert_id": alert_id, "code": "phq9_item9", "level": "acute"}
+
+
+async def test_audit_helper(dbs: None) -> None:
+    async with session.get_clinical_session() as s:
+        await audit(s, actor="system", action="crisis_acute", token=TOKEN, details=None, ts=T0)
+    async with session.get_clinical_session() as s:
+        entry = await s.scalar(select(AuditLog))
+        assert entry is not None and entry.action == "crisis_acute"
+
+
+async def test_dialog_state_messages_clock(dbs: None) -> None:
+    async with session.get_clinical_session() as s:
+        assert await repo.get_dialog_state(s, TOKEN) is None
+        await repo.set_dialog_state(s, token=TOKEN, state="checkin_mood", data={"step": 1}, now=T0)
+        await repo.add_message(
+            s, token=TOKEN, direction="in", text=ScrubbedText("ок"), ts=T0, fsm_state="idle"
+        )
+        with pytest.raises(ValueError):
+            await repo.add_message(s, token=TOKEN, direction="x", text=ScrubbedText("ок"), ts=T0)
+        assert await repo.load_clock_offset(s) == 0.0
+        await repo.save_clock_offset(s, 86400.0)
+    async with session.get_clinical_session() as s:
+        await repo.set_dialog_state(s, token=TOKEN, state="checkin_med", data={"step": 2}, now=T0)
+        await repo.save_clock_offset(s, 2 * 86400.0)
+    async with session.get_clinical_session() as s:
+        state = await repo.get_dialog_state(s, TOKEN)
+        assert state is not None and (state.state, state.data) == ("checkin_med", {"step": 2})
+        assert await repo.load_clock_offset(s) == 2 * 86400.0
+
+
+async def test_naive_datetime_rejected(dbs: None) -> None:
+    with pytest.raises(Exception, match="naive datetime"):
+        async with session.get_clinical_session() as s:
+            await repo.add_alert(
+                s, token=TOKEN, level=AlertLevel.INFO, code="x", reason="x", ts=datetime(2026, 1, 1)
+            )
+
+
+async def test_session_rolls_back_on_error(dbs: None) -> None:
+    with pytest.raises(RuntimeError):
+        async with session.get_clinical_session() as s:
+            await repo.upsert_profile(s, token=TOKEN, doctor_id="d1")
+            raise RuntimeError
+    async with session.get_clinical_session() as s:
+        assert await repo.get_profile(s, TOKEN) is None
+
+
+# --- guardrail #5: clinical code never touches identity ----------------------------------------------
+
+CLINICAL_MODULES = ["clinical.py", "repo.py", "audit.py", "session.py", "types.py"]
+
+
+def _imported_modules(path: Path) -> set[str]:
+    tree = ast.parse(path.read_text(encoding="utf-8"))
+    names: set[str] = set()
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Import):
+            names.update(alias.name for alias in node.names)
+        elif isinstance(node, ast.ImportFrom):
+            names.add(node.module or "")
+            names.update(f"{node.module}.{alias.name}" for alias in node.names)
+    return names
+
+
+@pytest.mark.parametrize("filename", CLINICAL_MODULES)
+def test_clinical_code_does_not_import_identity(filename: str) -> None:
+    imported = _imported_modules(DB_DIR / filename)
+    assert not [m for m in imported if "identity" in m], f"{filename} imports identity: {imported}"
+
+
+def test_clinical_import_graph_excludes_identity() -> None:
+    code = "import sys, app.db.repo, app.db.audit; print('app.db.identity' in sys.modules)"
+    out = subprocess.run(
+        [sys.executable, "-c", code], capture_output=True, text=True, check=True, cwd=DB_DIR.parents[1]
+    )
+    assert out.stdout.strip() == "False"
+
+
+def test_metadata_is_separate() -> None:
+    assert not set(IdentityBase.metadata.tables) & set(ClinicalBase.metadata.tables)
+    assert "audit_log" in ClinicalBase.metadata.tables