feat(backend): data model — identity / clinical / audit, repositories, create_all

Two DeclarativeBases on two engines (ADR-003): identity holds direct
identifiers, clinical holds token-keyed pseudonymised data plus audit_log.
Portable types (JSONB variant, tz-aware UTC timestamps rejecting naive
values) so tests run on SQLite. init_db() runs create_all at startup.
Guardrail #5 checked by AST and import-graph tests.

Refs #6

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

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