dashboard/console/__init__.py

@@ -0,0 +1,5 @@
+"""Demo console building blocks (#17): HTTP API helpers, scenario parser, playback runner, Streamlit widgets.
+
+Pages import this package as `console` (Streamlit puts `dashboard/` on sys.path). Modules other than
+`widgets` / `chat_ui` are free of Streamlit so they can be unit-tested.
+"""

dashboard/console/api.py

@@ -0,0 +1,185 @@
+"""Thin synchronous HTTP client for the demo console (frozen API contract, `app/api/schemas.py`).
+
+The console never imports backend code: it talks to `API_BASE_URL` only. Responses are kept as plain dicts —
+the console reads a handful of fields and must survive a backend that still serves fixtures (#12).
+Every failure is raised as `ApiError` with a short Russian message, so pages can show it instead of crashing.
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Any, Protocol
+
+import httpx
+
+DEFAULT_DOCTOR = "doc-smirnova"
+DEFAULT_TOKEN = "tok-andrey"
+
+
+def api_base_url() -> str:
+    return os.environ.get("API_BASE_URL", "http://localhost:8000").rstrip("/")
+
+
+def doctor_id() -> str:
+    return os.environ.get("DEMO_DOCTOR_ID", DEFAULT_DOCTOR)
+
+
+def fallback_token() -> str:
+    return os.environ.get("DEMO_PERSONA_TOKEN", DEFAULT_TOKEN)
+
+
+class ApiError(RuntimeError):
+    """Any API failure (network, HTTP status, bad JSON) with a message fit for the presenter."""
+
+
+class DemoApi(Protocol):
+    """What the playback runner needs; `ConsoleApi` implements it, tests use a fake."""
+
+    def clock(self) -> dict[str, Any]: ...
+
+    def advance(self, days: float = 0, hours: float = 0) -> dict[str, Any]: ...
+
+    def send_event(
+        self,
+        token: str,
+        type_: str,
+        *,
+        text: str | None = None,
+        button: str | None = None,
+        params: dict[str, Any] | None = None,
+    ) -> list[dict[str, Any]]: ...
+
+    def outbox(self, token: str) -> list[dict[str, Any]]: ...
+
+    def patients(self, doctor: str) -> list[dict[str, Any]]: ...
+
+
+class ConsoleApi:
+    """Client for `/api/v1`. Short timeouts everywhere except clock moves, which replay the scheduler."""
+
+    def __init__(self, base_url: str | None = None, *, timeout: float = 5.0, client: httpx.Client | None = None):
+        self.base_url = (base_url or api_base_url()).rstrip("/")
+        self._timeout = timeout
+        self._client = client or httpx.Client(timeout=timeout)
+
+    # ------------------------------------------------------------------ plumbing
+    def _request(self, method: str, path: str, *, timeout: float | None = None, **kwargs: Any) -> Any:
+        url = f"{self.base_url}{path}"
+        try:
+            resp = self._client.request(method, url, timeout=timeout or self._timeout, **kwargs)
+        except httpx.TimeoutException as exc:
+            raise ApiError(f"{method} {path}: API не ответил вовремя") from exc
+        except httpx.HTTPError as exc:
+            raise ApiError(f"{method} {path}: API недоступен ({type(exc).__name__})") from exc
+        if resp.status_code >= 400:
+            raise ApiError(f"{method} {path}: HTTP {resp.status_code}")
+        try:
+            return resp.json()
+        except ValueError as exc:
+            raise ApiError(f"{method} {path}: ответ не JSON") from exc
+
+    def close(self) -> None:
+        self._client.close()
+
+    # ------------------------------------------------------------------ service
+    def health(self) -> bool:
+        try:
+            return self._request("GET", "/health", timeout=2).get("status") == "ok"
+        except (ApiError, AttributeError):
+            return False
+
+    # ------------------------------------------------------------------ demo controls (ADR-005)
+    def personas(self) -> list[dict[str, Any]]:
+        return list(self._request("GET", "/api/v1/demo/personas"))
+
+    def clock(self) -> dict[str, Any]:
+        return dict(self._request("GET", "/api/v1/demo/clock"))
+
+    def advance(self, days: float = 0, hours: float = 0) -> dict[str, Any]:
+        # The backend replays the scheduler up to the new time inside this call: allow it time.
+        body = {"days": days, "hours": hours}
+        return dict(self._request("POST", "/api/v1/demo/clock/advance", json=body, timeout=60))
+
+    def reset(self) -> dict[str, Any]:
+        return dict(self._request("POST", "/api/v1/demo/reset", timeout=60))
+
+    # ------------------------------------------------------------------ patient dialog
+    def send_event(
+        self,
+        token: str,
+        type_: str,
+        *,
+        text: str | None = None,
+        button: str | None = None,
+        params: dict[str, Any] | None = None,
+    ) -> list[dict[str, Any]]:
+        body: dict[str, Any] = {"type": type_, "params": params or {}}
+        if text is not None:
+            body["text"] = text
+        if button is not None:
+            body["button"] = button
+        data = self._request("POST", f"/api/v1/dialog/{token}/events", json=body, timeout=30)
+        return list(data.get("replies", []))
+
+    def outbox(self, token: str) -> list[dict[str, Any]]:
+        return list(self._request("GET", f"/api/v1/dialog/{token}/outbox").get("replies", []))
+
+    def history(self, token: str, limit: int = 100) -> list[dict[str, Any]]:
+        return list(self._request("GET", f"/api/v1/dialog/{token}/history", params={"limit": limit}))
+
+    # ------------------------------------------------------------------ doctor side
+    def patients(self, doctor: str) -> list[dict[str, Any]]:
+        return list(self._request("GET", f"/api/v1/doctors/{doctor}/patients"))
+
+    def alerts(self, doctor: str, since_id: int = -1, open_only: bool = False) -> list[dict[str, Any]]:
+        params = {"since_id": since_id, "open_only": str(open_only).lower()}
+        return list(self._request("GET", f"/api/v1/doctors/{doctor}/alerts", params=params))
+
+    def connect(self, body: dict[str, Any]) -> dict[str, Any]:
+        """`POST /patients/connect` (ConnectPatientIn) → {token, code, deep_link}."""
+        return dict(self._request("POST", "/api/v1/patients/connect", json=body, timeout=30))
+
+    def ack(self, alert_id: int, doctor: str) -> dict[str, Any]:
+        return dict(self._request("POST", f"/api/v1/alerts/{alert_id}/ack", json={"doctor_id": doctor}))
+
+
+def personas_or_fallback(api: ConsoleApi) -> tuple[list[dict[str, Any]], str | None]:
+    """Demo personas for the selector, degrading step by step so the console always has someone to talk as.
+
+    1. `GET /demo/personas` (patients with a scenario file);
+    2. the demo doctor's patient list (real tokens; no scenario file yet, e.g. before #18 lands);
+    3. a single persona with the token from `DEMO_PERSONA_TOKEN` (fixtures use `tok-andrey`).
+    Returns the personas and a note for the presenter when a fallback was used.
+    """
+    try:
+        personas = [p for p in api.personas() if isinstance(p, dict) and p.get("token")]
+        if personas:
+            return personas, None
+        error = "API не вернул персон со сценарием"
+    except ApiError as exc:
+        error = str(exc)
+    try:
+        patients = [p for p in api.patients(doctor_id()) if isinstance(p, dict) and p.get("token")]
+    except ApiError:
+        patients = []
+    if patients:
+        # The demo hero first: the scenario file is guessed from the name later (`default_scenario_path`).
+        patients.sort(key=lambda p: not str(p.get("display_name", "")).startswith("Андрей"))
+        personas = [
+            {
+                "token": p["token"],
+                "display_name": p.get("display_name") or p["token"],
+                "therapy_day": p.get("therapy_day", 0),
+                "scenario": None,
+            }
+            for p in patients
+        ]
+        return personas, f"{error}; показаны пациенты врача"
+    token = fallback_token()
+    fallback = {
+        "token": token,
+        "display_name": "Андрей С." if token == DEFAULT_TOKEN else token,
+        "therapy_day": 0,
+        "scenario": None,
+    }
+    return [fallback], f"{error}; используется токен из окружения"

dashboard/console/chatlog.py

@@ -0,0 +1,81 @@
+"""In-memory transcript of the patient chat shown by the console (kept in Streamlit session state).
+
+Entries are plain dicts: `{"dir": "out", "text", "kind", "buttons"}` for bot replies (the `Reply` contract)
+and `{"dir": "in", "text", "pressed": bool}` for what the patient sent.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterable
+from typing import Any
+
+MAX_ENTRIES = 400
+
+
+def normalize_reply(raw: dict[str, Any]) -> dict[str, Any]:
+    """Coerce an API reply into a safe shape: missing fields get defaults, malformed buttons are dropped."""
+    rows: list[list[dict[str, str]]] = []
+    for row in raw.get("buttons") or []:
+        if not isinstance(row, list):
+            continue
+        clean = [
+            {"label": str(b.get("label", "")), "data": str(b.get("data", ""))}
+            for b in row
+            if isinstance(b, dict) and b.get("data") is not None
+        ]
+        if clean:
+            rows.append(clean)
+    return {
+        "dir": "out",
+        "text": str(raw.get("text") or ""),
+        "kind": str(raw.get("kind") or "normal"),
+        "buttons": rows,
+    }
+
+
+class ChatLog:
+    """Wraps a list owned by the caller (session state), so every page sees the same transcript."""
+
+    def __init__(self, entries: list[dict[str, Any]] | None = None):
+        self.entries: list[dict[str, Any]] = entries if entries is not None else []
+
+    def _trim(self) -> None:
+        if len(self.entries) > MAX_ENTRIES:
+            del self.entries[: len(self.entries) - MAX_ENTRIES]
+
+    def add_in(self, text: str, *, pressed: bool = False) -> None:
+        self.entries.append({"dir": "in", "text": text, "pressed": pressed})
+        self._trim()
+
+    def add_replies(self, replies: Iterable[dict[str, Any]]) -> list[dict[str, Any]]:
+        added = [normalize_reply(r) for r in replies if isinstance(r, dict)]
+        self.entries.extend(added)
+        self._trim()
+        return added
+
+    def add_history(self, messages: Iterable[dict[str, Any]]) -> None:
+        """Prefill from `GET /dialog/{token}/history` (MessageOut: no buttons)."""
+        for m in messages:
+            if not isinstance(m, dict):
+                continue
+            if m.get("direction") == "in":
+                self.entries.append({"dir": "in", "text": str(m.get("text", "")), "pressed": False})
+            else:
+                self.entries.append(normalize_reply({"text": m.get("text"), "kind": m.get("kind")}))
+        self._trim()
+
+    def recent_replies(self, n: int = 10) -> list[dict[str, Any]]:
+        """Last `n` bot replies, oldest first."""
+        outs = [e for e in self.entries if e.get("dir") == "out"]
+        return outs[-n:] if n > 0 else []
+
+    def last_buttons_index(self) -> int | None:
+        """Index of the newest bot reply that carries buttons — only its keyboard is clickable in the web chat."""
+        for i in range(len(self.entries) - 1, -1, -1):
+            e = self.entries[i]
+            if e.get("dir") == "out" and e.get("buttons"):
+                return i
+        return None
+
+    def clear(self) -> None:
+        self.entries.clear()

dashboard/console/playback.py

@@ -0,0 +1,312 @@
+"""Scenario playback over the HTTP API (#17): runs `data/synthetic/<persona>.jsonl` step by step.
+
+Time handling. A scenario day is a calendar date on the demo clock. When day N is played, its date is
+derived from the last played day M: date(N) = date(M) + (N − M); with nothing played yet (or if that date
+is already behind the clock, e.g. after manual «Следующий день») `advance_to HH:MM` goes to the next
+occurrence of HH:MM. The clock only moves forward (`/demo/clock/advance` replays the scheduler), so a target
+already in the past is skipped with a warning instead of failing.
+
+Failure handling. Every step is wrapped: an error stops playback, the position is kept in `PlaybackState`
+and the page offers «повторить шаг» / «пропустить шаг» / «продолжить вручную». The demo never crashes.
+"""
+
+from __future__ import annotations
+
+import time as _time
+from collections.abc import Callable, Iterator, Sequence
+from dataclasses import dataclass, field
+from datetime import date, datetime, time, timedelta
+from typing import Any, Literal
+
+from .api import DEFAULT_DOCTOR, ApiError, DemoApi
+from .chatlog import ChatLog
+from .scenario import Action, DayScript, Scenario, parse_hhmm
+
+# Virtual time keeps ticking between steps: a target a few minutes behind "now" counts as reached.
+TOLERANCE = timedelta(minutes=5)
+PRESS_WINDOW = 10
+
+StepStatus = Literal["ok", "warn", "fail", "info"]
+
+
+# ------------------------------------------------------------------------------------------------ pure helpers
+def _norm(s: str) -> str:
+    return " ".join(s.casefold().replace("ё", "е").split())
+
+
+def resolve_press(replies: Sequence[dict[str, Any]], label: str, window: int = PRESS_WINDOW) -> tuple[str, str] | None:
+    """Find the button to press for a scenario `press` step.
+
+    Searches the last `window` replies newest first; inside a reply an exact (case/ё-insensitive) label match
+    wins over a substring match. Returns `(data, label)` or None. Button `data` is dynamic (dose ids), so
+    scenarios address buttons by label.
+    """
+    needle = _norm(label)
+    if not needle:
+        return None
+    for reply in reversed(list(replies)[-window:] if window > 0 else []):
+        buttons = [b for row in reply.get("buttons") or [] for b in row if isinstance(b, dict)]
+        for exact in (True, False):
+            for b in buttons:
+                text = _norm(str(b.get("label", "")))
+                if (text == needle) if exact else (needle in text):
+                    return str(b.get("data", "")), str(b.get("label", ""))
+    return None
+
+
+def parse_clock_now(clock: dict[str, Any]) -> datetime:
+    """`ClockOut.now` (ISO, tz-aware) → datetime; raises ValueError on garbage."""
+    raw = clock.get("now")
+    if not isinstance(raw, str):
+        raise ValueError("часы API не вернули поле now")
+    return datetime.fromisoformat(raw.replace("Z", "+00:00"))
+
+
+def target_datetime(now: datetime, hhmm: str, day_date: date | None) -> datetime:
+    """Moment to move the clock to for `advance_to HH:MM` (see module docstring for the rules)."""
+    h, m = parse_hhmm(hhmm)
+    if day_date is None:
+        cand = now.replace(hour=h, minute=m, second=0, microsecond=0)
+        if cand < now - TOLERANCE:
+            cand += timedelta(days=1)
+        return cand
+    return datetime.combine(day_date, time(h, m), tzinfo=now.tzinfo)
+
+
+def hours_until(now: datetime, target: datetime) -> float | None:
+    """Hours to advance (≥ 0), 0 within the tolerance window, None if the target is clearly in the past."""
+    delta = target - now
+    if delta >= timedelta(0):
+        return delta.total_seconds() / 3600
+    if delta >= -TOLERANCE:
+        return 0.0
+    return None
+
+
+def planned_date(dates: dict[int, date], day: int, today: date) -> date | None:
+    """Calendar date for scenario `day` from already played days, or None to use "next occurrence" rules."""
+    if day in dates:
+        cand = dates[day]
+    else:
+        earlier = [d for d in dates if d < day]
+        if not earlier:
+            return None
+        prev = max(earlier)
+        cand = dates[prev] + timedelta(days=day - prev)
+    return cand if cand >= today else None
+
+
+# ------------------------------------------------------------------------------------------------ runner
+@dataclass
+class StepLog:
+    day: int
+    index: int  # 0-based step index inside the day; -1 for day-level messages
+    status: StepStatus
+    message: str
+    action: Action | None = None
+
+    def render(self) -> str:
+        icon = {"ok": "✅", "warn": "⚠️", "fail": "❌", "info": "ℹ️"}[self.status]
+        step = f"шаг {self.index + 1}: " if self.index >= 0 else ""
+        return f"{icon} день {self.day}, {step}{self.message}"
+
+
+@dataclass
+class Resume:
+    day: int
+    step: int
+    until: int
+
+
+@dataclass
+class PlaybackState:
+    """Per-persona playback memory kept in session state; survives an interrupted Streamlit run."""
+
+    dates: dict[int, date] = field(default_factory=dict)  # scenario day → calendar date it was played on
+    done: list[int] = field(default_factory=list)  # fully played days
+    failed: Resume | None = None  # where playback stopped on an error
+    position: Resume | None = None  # step currently being executed (left set if the run was interrupted)
+    log: list[str] = field(default_factory=list)
+
+    @property
+    def last_done(self) -> int | None:
+        return max(self.done) if self.done else None
+
+    def reset(self) -> None:
+        self.dates.clear()
+        self.done.clear()
+        self.failed = None
+        self.position = None
+        self.log.clear()
+
+
+class StepFailed(Exception):
+    pass
+
+
+class PlaybackRunner:
+    def __init__(
+        self,
+        api: DemoApi,
+        token: str,
+        chat: ChatLog,
+        state: PlaybackState,
+        *,
+        doctor: str = DEFAULT_DOCTOR,
+        delay: float = 0.0,
+        press_retries: int = 3,
+        retry_wait: float = 0.7,
+        sleep: Callable[[float], None] = _time.sleep,
+    ):
+        self.api = api
+        self.token = token
+        self.chat = chat
+        self.state = state
+        self.doctor = doctor
+        self.delay = delay
+        self.press_retries = press_retries
+        self.retry_wait = retry_wait
+        self.sleep = sleep
+
+    # ---------------------------------------------------------------- helpers
+    def drain(self) -> list[dict[str, Any]]:
+        """Pull system-initiated replies (reminders, check-ins) into the chat."""
+        return self.chat.add_replies(self.api.outbox(self.token))
+
+    def _now(self) -> datetime:
+        return parse_clock_now(self.api.clock())
+
+    @staticmethod
+    def _preview(replies: list[dict[str, Any]]) -> str:
+        if not replies:
+            return ""
+        text = replies[-1].get("text", "").replace("\n", " ")
+        more = f" (+{len(replies) - 1})" if len(replies) > 1 else ""
+        return f" → «{text[:70]}{'…' if len(text) > 70 else ''}»{more}"
+
+    def next_day(self, scenario: Scenario) -> DayScript | None:
+        last = self.state.last_done
+        return next((d for d in scenario.days if last is None or d.day > last), None)
+
+    # ---------------------------------------------------------------- one step
+    def _step(self, day: DayScript, action: Action, day_date: list[date | None]) -> tuple[StepStatus, str]:
+        if action.kind == "advance_to":
+            now = self._now()
+            target = target_datetime(now, action.value, day_date[0])
+            if day_date[0] is None:
+                day_date[0] = target.date()
+                self.state.dates[day.day] = target.date()
+            hours = hours_until(now, target)
+            if hours is None:
+                return "warn", f"{action.describe()}: время уже прошло ({now:%d.%m %H:%M}), пропускаю"
+            if hours > 0:
+                self.api.advance(hours=hours)
+            got = self.drain()
+            return "ok", f"{action.describe()} ({target:%d.%m %H:%M}){self._preview(got)}"
+
+        if action.kind == "press":
+            found = resolve_press(self.chat.recent_replies(PRESS_WINDOW), action.value)
+            for _ in range(self.press_retries):
+                if found is not None:
+                    break
+                # The reply may still be on its way (scheduler tick): wait and drain the outbox again.
+                self.sleep(self.retry_wait)
+                self.drain()
+                found = resolve_press(self.chat.recent_replies(PRESS_WINDOW), action.value)
+            if found is None:
+                raise StepFailed(f"кнопка «{action.value}» не найдена в последних {PRESS_WINDOW} сообщениях")
+            data, label = found
+            self.chat.add_in(label, pressed=True)
+            got = self.chat.add_replies(self.api.send_event(self.token, "button", button=data))
+            got += self.drain()
+            return "ok", f"{action.describe()}{self._preview(got)}"
+
+        if action.kind == "text":
+            self.chat.add_in(action.value)
+            got = self.chat.add_replies(self.api.send_event(self.token, "text", text=action.value))
+            got += self.drain()
+            return "ok", f"{action.describe()}{self._preview(got)}"
+
+        if action.kind == "start":
+            params = {"code": action.value} if action.value else {}
+            self.chat.add_in("/start", pressed=True)
+            got = self.chat.add_replies(self.api.send_event(self.token, "start", params=params))
+            got += self.drain()
+            return "ok", f"{action.describe()}{self._preview(got)}"
+
+        # expect_color is a soft check: a mismatch is shown to the presenter but never stops the demo.
+        try:
+            patients = self.api.patients(self.doctor)
+        except ApiError as exc:
+            return "warn", f"{action.describe()}: не удалось проверить ({exc})"
+        me = next((p for p in patients if isinstance(p, dict) and p.get("token") == self.token), None)
+        if me is None:
+            return "warn", f"{action.describe()}: пациента нет в списке врача"
+        actual = me.get("status_color")
+        if actual == action.value:
+            return "ok", f"{action.describe()}: совпало"
+        return "warn", f"{action.describe()}: у врача сейчас {actual}"
+
+    # ---------------------------------------------------------------- days
+    def run(self, scenario: Scenario, from_day: int, from_step: int, until_day: int) -> Iterator[StepLog]:
+        """Play days `from_day..until_day` (those present in the scenario), the first one from `from_step`."""
+        days = [d for d in scenario.days if from_day <= d.day <= until_day]
+        if not days:
+            yield StepLog(from_day, -1, "info", "в сценарии нет дней для проигрывания")
+            return
+        self.state.failed = None
+        for n, day in enumerate(days):
+            start = from_step if n == 0 else 0
+            try:
+                today = self._now().date()
+            except (ApiError, ValueError) as exc:
+                self._fail(day.day, max(start, 0), until_day)
+                yield StepLog(day.day, -1, "fail", f"часы API недоступны: {exc}")
+                return
+            day_date: list[date | None] = [planned_date(self.state.dates, day.day, today)]
+            if day_date[0] is not None:
+                self.state.dates[day.day] = day_date[0]
+            yield StepLog(day.day, -1, "info", f"начинаю день {day.day}" + (f": {day.note}" if day.note else ""))
+            for i in range(start, len(day.actions)):
+                action = day.actions[i]
+                self.state.position = Resume(day.day, i, until_day)
+                try:
+                    status, message = self._step(day, action, day_date)
+                except Exception as exc:  # robustness first: any error pauses playback, never crashes the page
+                    self._fail(day.day, i, until_day)
+                    reason = str(exc) or type(exc).__name__
+                    yield StepLog(day.day, i, "fail", f"{action.describe()}: {reason}", action)
+                    return
+                yield StepLog(day.day, i, status, message, action)
+                if self.delay > 0 and i < len(day.actions) - 1:
+                    self.sleep(self.delay)
+            if day.day not in self.state.dates:
+                self.state.dates[day.day] = today
+            if day.day not in self.state.done:
+                self.state.done.append(day.day)
+            self.state.position = None
+            yield StepLog(day.day, -1, "ok", f"день {day.day} проигран")
+            if self.delay > 0 and n < len(days) - 1:
+                self.sleep(self.delay)
+
+    def _fail(self, day: int, step: int, until: int) -> None:
+        self.state.failed = Resume(day, step, until)
+        self.state.position = None
+
+    def play_day(self, scenario: Scenario, day: int) -> Iterator[StepLog]:
+        return self.run(scenario, day, 0, day)
+
+    def scroll_to(self, scenario: Scenario, day: int) -> Iterator[StepLog]:
+        """Play every not yet played day up to and including `day`."""
+        nxt = self.next_day(scenario)
+        if nxt is None or nxt.day > day:
+            return iter([StepLog(day, -1, "info", f"дни до {day} уже проиграны")])
+        return self.run(scenario, nxt.day, 0, day)
+
+    def resume(self, scenario: Scenario, *, skip: bool) -> Iterator[StepLog]:
+        """Continue after a failure (or an interrupted run): retry the failed step or skip it."""
+        point = self.state.failed or self.state.position
+        if point is None:
+            return iter([StepLog(0, -1, "info", "нечего продолжать")])
+        step = point.step + (1 if skip else 0)
+        return self.run(scenario, point.day, step, point.until)

dashboard/console/playback_ui.py

@@ -0,0 +1,106 @@
+"""Scenario playback panel of the demo console (#17): day list, «Проиграть день N», «Прокрутить до дня N»."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import streamlit as st
+
+from . import widgets as w
+from .playback import PlaybackState
+from .scenario import Scenario
+
+
+def _forget_upload(token: str) -> None:
+    st.session_state.pop(f"console_scenario_text:{token}", None)
+    st.session_state.pop(f"console_up:{token}", None)  # resets the uploader widget too
+
+
+def _scenario_source_controls(persona: dict[str, Any], source: str) -> None:
+    token = persona["token"]
+    with st.expander(f"Файл сценария: {source}"):
+        st.text_input(
+            "Путь в репозитории",
+            key=f"console_scenario_path:{token}",
+            placeholder=str(persona.get("scenario") or "data/synthetic/andrey.jsonl"),
+        )
+        upload = st.file_uploader("…или загрузить .jsonl", type=["jsonl", "json", "txt"], key=f"console_up:{token}")
+        if upload is not None:
+            try:
+                st.session_state[f"console_scenario_text:{token}"] = upload.getvalue().decode("utf-8")
+            except UnicodeDecodeError:
+                st.error("Файл не в UTF-8")
+        if st.session_state.get(f"console_scenario_text:{token}"):
+            st.button("Забыть загруженный файл", on_click=_forget_upload, args=(token,))
+
+
+def _day_table(scenario: Scenario, state: PlaybackState, next_day: int | None) -> None:
+    rows = []
+    for d in scenario.days:
+        mark = "✅" if d.day in state.done else ("▶️" if d.day == next_day else "")
+        rows.append({"": mark, "День": d.day, "Шагов": len(d.actions), "Подсказка ведущему": d.note})
+    st.dataframe(rows, hide_index=True, width="stretch", height=min(38 + 35 * len(rows), 330))
+
+
+def _clear_failure(state: PlaybackState) -> None:
+    state.failed = None
+    state.position = None
+
+
+def render_playback_panel(persona: dict[str, Any]) -> None:
+    token = persona["token"]
+    scenario, source = w.scenario_for(persona)
+    state = w.playback_state(token)
+    st.subheader("🎬 Сценарий демо")
+    _scenario_source_controls(persona, source)
+    if scenario.errors:
+        with st.expander(f"⚠️ Замечания к сценарию: {len(scenario.errors)}", expanded=not scenario.days):
+            for err in scenario.errors:
+                st.write(f"- {err}")
+    if not scenario.days:
+        st.info("Сценарий не загружен. Демо можно вести вручную: чат слева и кнопки часов сверху.")
+        return
+
+    delay = st.slider("Пауза между шагами, с", 0.0, 3.0, 1.0, 0.25, key=f"console_delay:{token}")
+    runner = w.make_runner(persona, delay=delay)
+    nxt = runner.next_day(scenario)
+    _day_table(scenario, state, nxt.day if nxt else None)
+
+    numbers = scenario.day_numbers
+    default = numbers.index(nxt.day) if nxt else len(numbers) - 1
+    day = st.selectbox("День", numbers, index=default, key=f"console_day:{token}", format_func=lambda n: f"день {n}")
+    c1, c2, c3 = st.columns(3)
+    action: tuple[str, Any] | None = None
+    if c1.button(f"▶ Проиграть день {day}", type="primary", width="stretch"):
+        action = ("Проигрываю день", runner.play_day(scenario, day))
+    if c2.button(f"⏭ Прокрутить до дня {day}", width="stretch", help="Быстро, без пауз: все непроигранные дни"):
+        runner.delay = 0.0
+        action = (f"Прокрутка до дня {day}", runner.scroll_to(scenario, day))
+    if c3.button(
+        f"▶ Следующий: день {nxt.day}" if nxt else "Все дни проиграны",
+        width="stretch",
+        disabled=nxt is None,
+    ):
+        assert nxt is not None
+        action = ("Проигрываю день", runner.play_day(scenario, nxt.day))
+
+    point = state.failed or state.position
+    if point is not None and action is None:
+        what = "остановлено на ошибке" if state.failed else "прервано"
+        st.warning(f"Проигрывание {what}: день {point.day}, шаг {point.step + 1}.")
+        r1, r2, r3 = st.columns(3)
+        if r1.button("🔁 Повторить шаг", width="stretch"):
+            action = ("Продолжаю", runner.resume(scenario, skip=False))
+        if r2.button("⏩ Пропустить шаг", width="stretch"):
+            action = ("Продолжаю", runner.resume(scenario, skip=True))
+        r3.button("✋ Продолжить вручную", width="stretch", on_click=_clear_failure, args=(state,))
+
+    if action is not None:
+        title, logs = action
+        w.execute(logs, state, title=title)
+        st.session_state["console_log_open"] = True
+        st.rerun()  # refresh the chat and the clock with what playback produced
+
+    if state.log:
+        with st.expander("Журнал последнего проигрывания", expanded=st.session_state.pop("console_log_open", False)):
+            st.text("\n".join(state.log[-80:]))

dashboard/console/scenario.py

@@ -0,0 +1,180 @@
+"""Demo scenario files `data/synthetic/<persona>.jsonl` (format agreed with #18).
+
+One JSON object per line: `{"day": <therapy day>, "note": "<presenter hint>", "actions": [...]}` where an action
+is exactly one of `{"advance_to": "HH:MM"}`, `{"press": "<label substring>"}`, `{"text": "<message>"}`,
+`{"expect_color": "green|yellow|red|grey"}`. The console also accepts `{"start": "<connect code or empty>"}`
+(sends a `start` event) so a scenario can open the dialog itself.
+
+The parser is lenient on purpose: a broken line or action is reported and skipped, the rest still plays —
+a typo in the scenario must not take the live demo down.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Literal
+
+ActionKind = Literal["advance_to", "press", "text", "expect_color", "start"]
+ACTION_KINDS: tuple[ActionKind, ...] = ("advance_to", "press", "text", "expect_color", "start")
+COLORS = ("green", "yellow", "red", "grey")
+_HHMM = re.compile(r"^([01]?\d|2[0-3]):([0-5]\d)$")
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+
+
+@dataclass(frozen=True)
+class Action:
+    kind: ActionKind
+    value: str
+
+    def describe(self) -> str:
+        """Short Russian description for the step log."""
+        return {
+            "advance_to": f"⏩ часы → {self.value}",
+            "press": f"🔘 нажать «{self.value}»",
+            "text": f"💬 написать «{self.value}»",
+            "expect_color": f"🎯 ожидаем цвет {self.value}",
+            "start": "🚀 начать диалог" + (f" (код {self.value})" if self.value else ""),
+        }[self.kind]
+
+
+@dataclass(frozen=True)
+class DayScript:
+    day: int
+    note: str
+    actions: tuple[Action, ...]
+
+
+@dataclass
+class Scenario:
+    days: list[DayScript] = field(default_factory=list)
+    errors: list[str] = field(default_factory=list)
+
+    def day(self, number: int) -> DayScript | None:
+        return next((d for d in self.days if d.day == number), None)
+
+    @property
+    def day_numbers(self) -> list[int]:
+        return [d.day for d in self.days]
+
+
+def parse_hhmm(value: str) -> tuple[int, int]:
+    """'8:05' / '08:05' → (8, 5); raises ValueError on anything else."""
+    m = _HHMM.match(value.strip())
+    if m is None:
+        raise ValueError(f"время должно быть в формате ЧЧ:ММ, получено «{value}»")
+    return int(m.group(1)), int(m.group(2))
+
+
+def parse_action(raw: Any) -> Action:
+    """Validate one action object; raises ValueError with a Russian message."""
+    if not isinstance(raw, dict) or len(raw) != 1:
+        raise ValueError(f"действие должно быть объектом с одним ключом, получено {raw!r}")
+    ((kind, value),) = raw.items()
+    if kind not in ACTION_KINDS:
+        raise ValueError(f"неизвестное действие «{kind}»")
+    if value is None and kind == "start":
+        value = ""
+    if not isinstance(value, str):
+        raise ValueError(f"значение «{kind}» должно быть строкой")
+    value = value.strip()
+    if kind == "advance_to":
+        h, m = parse_hhmm(value)
+        value = f"{h:02d}:{m:02d}"
+    elif kind == "expect_color":
+        value = value.lower()
+        if value not in COLORS:
+            raise ValueError(f"цвет должен быть одним из {', '.join(COLORS)}, получено «{value}»")
+    elif kind in ("press", "text") and not value:
+        raise ValueError(f"пустое значение «{kind}»")
+    return Action(kind, value)  # type: ignore[arg-type]
+
+
+def parse_scenario(text: str) -> Scenario:
+    """Parse JSONL text. Days are returned sorted; duplicates keep the first occurrence."""
+    scenario = Scenario()
+    seen: set[int] = set()
+    for lineno, line in enumerate(text.splitlines(), start=1):
+        line = line.strip()
+        if not line or line.startswith("//") or line.startswith("#"):
+            continue
+        where = f"строка {lineno}"
+        try:
+            obj = json.loads(line)
+        except json.JSONDecodeError as exc:
+            scenario.errors.append(f"{where}: не JSON ({exc.msg})")
+            continue
+        if not isinstance(obj, dict):
+            scenario.errors.append(f"{where}: ожидается объект")
+            continue
+        day = obj.get("day")
+        if isinstance(day, bool) or not isinstance(day, int) or day < 0:
+            scenario.errors.append(f"{where}: поле day должно быть целым ≥ 0")
+            continue
+        if day in seen:
+            scenario.errors.append(f"{where}: день {day} повторяется, строка пропущена")
+            continue
+        raw_actions = obj.get("actions", [])
+        if not isinstance(raw_actions, list):
+            scenario.errors.append(f"{where}: actions должно быть списком")
+            continue
+        actions: list[Action] = []
+        for i, raw in enumerate(raw_actions, start=1):
+            try:
+                actions.append(parse_action(raw))
+            except ValueError as exc:
+                scenario.errors.append(f"{where}, день {day}, действие {i}: {exc} — пропущено")
+        note = obj.get("note", "")
+        seen.add(day)
+        scenario.days.append(DayScript(day=day, note=str(note) if note is not None else "", actions=tuple(actions)))
+    scenario.days.sort(key=lambda d: d.day)
+    return scenario
+
+
+def resolve_path(path: str | Path, root: Path = REPO_ROOT) -> Path:
+    p = Path(path)
+    return p if p.is_absolute() else root / p
+
+
+def load_scenario(path: str | Path, root: Path = REPO_ROOT) -> Scenario:
+    """Read and parse a scenario file; a missing/unreadable file yields an empty scenario with an error."""
+    full = resolve_path(path, root)
+    try:
+        text = full.read_text(encoding="utf-8")
+    except FileNotFoundError:
+        return Scenario(errors=[f"файл сценария не найден: {path}"])
+    except OSError as exc:
+        return Scenario(errors=[f"не удалось прочитать {path}: {type(exc).__name__}"])
+    return parse_scenario(text)
+
+
+_TRANSLIT = {
+    "а": "a", "б": "b", "в": "v", "г": "g", "д": "d", "е": "e", "ё": "e", "ж": "zh", "з": "z", "и": "i",
+    "й": "y", "к": "k", "л": "l", "м": "m", "н": "n", "о": "o", "п": "p", "р": "r", "с": "s", "т": "t",
+    "у": "u", "ф": "f", "х": "kh", "ц": "ts", "ч": "ch", "ш": "sh", "щ": "shch", "ъ": "", "ы": "y", "ь": "",
+    "э": "e", "ю": "yu", "я": "ya",
+}  # fmt: skip
+
+
+def persona_slug(display_name: str) -> str:
+    """'Андрей С.' → 'andrey': the backend names scenario files by the transliterated first name (#12)."""
+    first = (display_name.split() or [""])[0].lower()
+    return "".join(_TRANSLIT.get(ch, ch if ch.isascii() and ch.isalnum() else "") for ch in first)
+
+
+def default_scenario_path(persona: dict[str, Any], root: Path = REPO_ROOT) -> str | None:
+    """Scenario path from the persona, else `data/synthetic/<slug>.jsonl` guessed from the name or `tok-<slug>`."""
+    if persona.get("scenario"):
+        return str(persona["scenario"])
+    token = str(persona.get("token", ""))
+    slugs = [persona_slug(str(persona.get("display_name", "")))]
+    if token.startswith("tok-"):
+        slugs.append(token.removeprefix("tok-"))
+    for slug in filter(None, slugs):
+        guess = f"data/synthetic/{slug}.jsonl"
+        if resolve_path(guess, root).exists():
+            return guess
+    return None

dashboard/console/widgets.py

@@ -0,0 +1,455 @@
+"""Streamlit widgets shared by the demo console and the split screen (#17).
+
+Robustness rules for the live demo: every API call goes through `ConsoleApi` (raises `ApiError` only),
+callbacks never raise — errors are queued as flash messages and shown on the next render — and the chat
+transcript / playback progress live in `st.session_state`, so both pages share them within one browser tab.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import os
+from datetime import datetime
+from typing import Any
+
+import streamlit as st
+
+from .api import ApiError, ConsoleApi, api_base_url, doctor_id, personas_or_fallback
+from .chatlog import ChatLog
+from .playback import PlaybackRunner, PlaybackState, StepLog, parse_clock_now, target_datetime
+from .scenario import Scenario, default_scenario_path, load_scenario, parse_scenario
+
+COLOR_EMOJI = {"red": "🔴", "yellow": "🟡", "green": "🟢", "grey": "⚪"}
+LEVEL_EMOJI = {"acute": "🔴", "high": "🟠", "moderate": "🟡", "info": "⚪"}
+WEEKDAYS = ("пн", "вт", "ср", "чт", "пт", "сб", "вс")
+POLL_SECONDS = 2
+
+
+# ------------------------------------------------------------------------------------------------ state
+@st.cache_resource
+def _api_for(base_url: str) -> ConsoleApi:
+    return ConsoleApi(base_url)
+
+
+def get_api() -> ConsoleApi:
+    return _api_for(api_base_url())
+
+
+def chat_for(token: str) -> ChatLog:
+    key = f"console_chat:{token}"
+    if key not in st.session_state:
+        entries: list[dict[str, Any]] = []
+        st.session_state[key] = entries
+        with contextlib.suppress(ApiError):  # history is a nice-to-have: the chat may start empty
+            ChatLog(entries).add_history(get_api().history(token))
+    return ChatLog(st.session_state[key])
+
+
+def playback_state(token: str) -> PlaybackState:
+    key = f"console_playback:{token}"
+    if key not in st.session_state:
+        st.session_state[key] = PlaybackState()
+    return st.session_state[key]
+
+
+def flash(message: str, level: str = "error") -> None:
+    """Report a problem from a callback: a toast now plus a persistent note shown by `show_flash`."""
+    icon = {"error": "❌", "warning": "⚠️", "success": "✅"}.get(level, "ℹ️")
+    # Outside a script run context (tests, threads) a toast may fail — the persistent note below is enough.
+    with contextlib.suppress(Exception):
+        st.toast(message, icon=icon)
+    st.session_state["console_last_note"] = (level, message)
+
+
+def show_flash(container: Any = None) -> None:
+    """Last error/notice of the session, kept until the next one (does not vanish with the 2 s poll)."""
+    note = st.session_state.get("console_last_note")
+    if not note:
+        return
+    level, message = note
+    where = container or st.sidebar
+    {"error": where.error, "warning": where.warning, "success": where.success}.get(level, where.info)(message)
+
+
+def forget_demo_state() -> None:
+    """After «Сброс демо»: drop transcripts, playback progress and the alert baseline of every persona."""
+    for key in list(st.session_state.keys()):
+        if str(key).startswith(("console_chat:", "console_playback:", "console_alerts_")):
+            del st.session_state[key]
+
+
+# ------------------------------------------------------------------------------------------------ persona
+@st.cache_data(ttl=15, show_spinner=False)
+def _personas(base_url: str) -> tuple[list[dict[str, Any]], str | None]:
+    return personas_or_fallback(get_api())
+
+
+def select_persona(container: Any = None) -> dict[str, Any]:
+    """Persona selector (sidebar by default); the choice is shared by all console pages."""
+    where = container or st.sidebar
+    personas, error = _personas(get_api().base_url)
+    if error:
+        where.caption(f"⚠️ {error}")
+    tokens = [p["token"] for p in personas]
+    current = st.session_state.get("console_token")
+    index = tokens.index(current) if current in tokens else 0
+    labels = {p["token"]: f"{p.get('display_name') or p['token']} ({p['token'][:12]})" for p in personas}
+    token = where.selectbox("Пациент демо", tokens, index=index, format_func=lambda t: labels.get(t, t))
+    st.session_state["console_token"] = token
+    return next(p for p in personas if p["token"] == token)
+
+
+# Synthetic demo hero (docs/product/01_demo_persona.md); the SNILS is fake and used only to derive the token.
+DEMO_HERO = {
+    "display_name": "Андрей С.",
+    "birth_year": 1992,
+    "sex": "m",
+    "snils": "000-000-017 17",
+    "drug": "сертралин",
+    "dose": "50 мг",
+    "times": ["08:00"],
+    "instructions": "1 раз в сутки утром после еды",
+    "modules": ["veteran", "depression"],
+}
+
+
+def _connect_hero() -> None:
+    name = st.session_state.get("console_hero_name") or DEMO_HERO["display_name"]
+    body = {**DEMO_HERO, "doctor_id": doctor_id(), "display_name": name}
+    try:
+        out = get_api().connect(body)
+    except ApiError as exc:
+        hint = " — пациент уже подключён, выберите его в списке" if "409" in str(exc) else ""
+        flash(f"Подключение не выполнено: {exc}{hint}", "warning" if hint else "error")
+        return
+    token = str(out.get("token", ""))
+    st.session_state["console_token"] = token
+    st.session_state[f"console_code:{token}"] = str(out.get("code", ""))
+    _personas.clear()
+    flash(f"Пациент подключён, код {out.get('code', '')}. Нажмите «Начать» в чате.", "success")
+
+
+def render_connect_hero(container: Any = None) -> None:
+    """Fallback when the demo seed is missing: connect the synthetic hero through the doctor API."""
+    where = container or st.sidebar
+    with where.expander("Подключить демо-пациента"):
+        st.caption("Если сид демо не загружен: врач подключает Андрея (синтетика) — как на приёме.")
+        st.text_input("Имя", value="Андрей С.", key="console_hero_name")
+        st.button("Подключить", key="console-connect", on_click=_connect_hero)
+
+
+# ------------------------------------------------------------------------------------------------ clock
+def _drain_quiet(api: ConsoleApi, token: str) -> None:
+    try:
+        chat_for(token).add_replies(api.outbox(token))
+    except ApiError as exc:
+        flash(f"Не удалось забрать входящие: {exc}", "warning")
+
+
+def _advance(token: str, days: float = 0, hours: float = 0) -> None:
+    api = get_api()
+    try:
+        api.advance(days=days, hours=hours)
+    except ApiError as exc:
+        flash(f"Часы не сдвинулись: {exc}")
+        return
+    _drain_quiet(api, token)
+
+
+def _advance_to(token: str, hhmm: str) -> None:
+    api = get_api()
+    try:
+        now = parse_clock_now(api.clock())
+        hours = (target_datetime(now, hhmm, None) - now).total_seconds() / 3600
+        if hours > 0:
+            api.advance(hours=hours)
+    except (ApiError, ValueError) as exc:
+        flash(f"Часы не сдвинулись: {exc}")
+        return
+    _drain_quiet(api, token)
+
+
+def _reset() -> None:
+    try:
+        get_api().reset()
+    except ApiError as exc:
+        flash(f"Сброс не выполнен: {exc}")
+        return
+    forget_demo_state()
+    _personas.clear()  # the seed may be recreated on reset
+    flash("Демо сброшено: часы вернулись к реальному времени, чат и прогресс сценария очищены.", "success")
+
+
+def _therapy_day(api: ConsoleApi, persona: dict[str, Any]) -> int | None:
+    try:
+        me = next((p for p in api.patients(doctor_id()) if p.get("token") == persona["token"]), None)
+    except ApiError:
+        me = None
+    day = (me or persona).get("therapy_day")
+    return day if isinstance(day, int) else None
+
+
+def format_now(now: datetime) -> str:
+    return f"{WEEKDAYS[now.weekday()]} {now:%d.%m.%Y} · {now:%H:%M}"
+
+
+def render_clock_bar(persona: dict[str, Any], *, key: str, with_reset: bool = True) -> None:
+    """Virtual date/time, therapy day and the time controls (ADR-005)."""
+    api = get_api()
+    token = persona["token"]
+    info, *buttons = st.columns([3.2, 1, 1.2, 1.2, 1.3, 1.2] if with_reset else [3.2, 1, 1.2, 1.2, 1.3])
+    try:
+        clock = api.clock()
+        now = parse_clock_now(clock)
+        day = _therapy_day(api, persona)
+        mode = "виртуальные часы" if clock.get("virtual") else "реальное время"
+        info.markdown(f"### 🕒 {format_now(now)}")
+        info.caption(f"День терапии: **{day if day is not None else '—'}** · {mode}")
+    except (ApiError, ValueError) as exc:
+        info.warning(f"Часы недоступны: {exc}")
+    buttons[0].button("+1 час", key=f"{key}-h1", on_click=_advance, args=(token,), kwargs={"hours": 1})
+    buttons[1].button("К 08:00 (приём)", key=f"{key}-08", on_click=_advance_to, args=(token, "08:00"))
+    buttons[2].button("К 20:00 (чек-ин)", key=f"{key}-20", on_click=_advance_to, args=(token, "20:00"))
+    buttons[3].button(
+        "Следующий день", key=f"{key}-d1", type="primary", on_click=_advance, args=(token,), kwargs={"days": 1}
+    )
+    if with_reset:
+        with buttons[4].popover("Сброс демо"):
+            st.write("Вернуть часы к реальному времени и очистить чат и прогресс сценария?")
+            st.button("Да, сбросить", key=f"{key}-reset", type="primary", on_click=_reset)
+
+
+# ------------------------------------------------------------------------------------------------ chat
+def _send(token: str, type_: str, *, text: str | None = None, button: str | None = None, label: str = "") -> None:
+    api = get_api()
+    chat = chat_for(token)
+    params: dict[str, Any] = {}
+    if type_ == "start":
+        code = (st.session_state.get(f"console_code:{token}") or "").strip()
+        params = {"code": code} if code else {}
+        chat.add_in("/start" + (f" {code}" if code else ""), pressed=True)
+    elif type_ == "button":
+        chat.add_in(label, pressed=True)
+    else:
+        chat.add_in(text or "")
+    try:
+        chat.add_replies(api.send_event(token, type_, text=text, button=button, params=params))
+    except ApiError as exc:
+        flash(f"Сообщение не доставлено: {exc}")
+        return
+    _drain_quiet(api, token)
+
+
+def _submit_text(token: str, key: str) -> None:
+    text = (st.session_state.get(key) or "").strip()
+    if text:
+        _send(token, "text", text=text)
+
+
+def _render_entry(entry: dict[str, Any]) -> None:
+    if entry.get("dir") == "in":
+        with st.chat_message("user", avatar="🧑"):
+            st.markdown(f"🔘 *{entry['text']}*" if entry.get("pressed") else entry["text"])
+        return
+    kind = entry.get("kind", "normal")
+    with st.chat_message("assistant", avatar="🤝"):
+        if kind == "crisis":
+            st.error(entry["text"], icon="🚨")
+        elif kind == "safety_plan":
+            with st.container(border=True):
+                st.markdown("**🛡️ План безопасности**")
+                st.markdown(entry["text"])
+        elif kind == "info":
+            st.caption(entry["text"])
+        else:
+            st.markdown(entry["text"])
+
+
+def render_messages(token: str, *, height: int, key: str) -> None:
+    """Transcript + the keyboard of the newest reply with buttons. Drains the outbox on every call."""
+    _drain_quiet(get_api(), token)
+    chat = chat_for(token)
+    box = st.container(height=height, border=True, autoscroll=True)
+    with box:
+        if not chat.entries:
+            st.caption("Сообщений пока нет. Нажмите «Начать» или сдвиньте часы к напоминанию.")
+        active = chat.last_buttons_index()
+        for i, entry in enumerate(chat.entries):
+            _render_entry(entry)
+            if i == active:
+                for r, row in enumerate(entry["buttons"]):
+                    cols = st.columns(len(row))
+                    for c, b in enumerate(row):
+                        cols[c].button(
+                            b["label"] or "…",
+                            key=f"{key}-b{i}-{r}-{c}",
+                            width="stretch",
+                            on_click=_send,
+                            args=(token, "button"),
+                            kwargs={"button": b["data"], "label": b["label"]},
+                        )
+
+
+@st.fragment(run_every=POLL_SECONDS)
+def _polled_messages(token: str, height: int, key: str) -> None:
+    render_messages(token, height=height, key=key)
+
+
+def render_chat(token: str, *, height: int = 520, key: str = "chat") -> None:
+    """Fallback patient web chat: polled transcript, text input, «Начать» with an optional connect code."""
+    _polled_messages(token, height, key)
+    input_key = f"{key}-input:{token}"
+    with st.form(f"{key}-form:{token}", clear_on_submit=True, border=False):
+        left, right = st.columns([5, 1])
+        left.text_input("Сообщение", key=input_key, label_visibility="collapsed", placeholder="Сообщение пациента…")
+        right.form_submit_button("➤", width="stretch", on_click=_submit_text, args=(token, input_key))
+    code = st.query_params.get("code")  # connect links from the doctor point to the console with ?code=
+    if code and not st.session_state.get(f"console_code:{token}"):
+        st.session_state[f"console_code:{token}"] = code
+    with st.expander("Начать диалог / код подключения", expanded=bool(code)):
+        st.text_input("Код подключения (необязательно)", key=f"console_code:{token}", placeholder="DEMO-0001")
+        c1, c2 = st.columns(2)
+        c1.button("🚀 Начать", key=f"{key}-start", on_click=_send, args=(token, "start"), width="stretch")
+        c2.button(
+            "Очистить окно чата",
+            key=f"{key}-clear",
+            on_click=lambda: chat_for(token).clear(),
+            width="stretch",
+        )
+
+
+def patient_web_url() -> str | None:
+    url = (os.environ.get("PATIENT_WEB_URL") or "").strip()
+    return url or None
+
+
+# ------------------------------------------------------------------------------------------------ doctor panel
+def _dismiss_banner(alert_id: int) -> None:
+    st.session_state.setdefault("console_alerts_dismissed", set()).add(alert_id)
+
+
+def _ack(alert_id: int) -> None:
+    try:
+        get_api().ack(alert_id, doctor_id())
+    except ApiError as exc:
+        flash(f"Алерт не подтверждён: {exc}")
+    _dismiss_banner(alert_id)
+
+
+def _fmt_ts(raw: Any) -> str:
+    try:
+        return datetime.fromisoformat(str(raw).replace("Z", "+00:00")).strftime("%d.%m %H:%M")
+    except ValueError:
+        return ""
+
+
+def render_doctor_panel(highlight_token: str | None, *, key: str = "doc") -> None:
+    """Compact doctor view for the projector: patients with colours and the live alerts feed."""
+    api = get_api()
+    doctor = doctor_id()
+    try:
+        alerts = [a for a in api.alerts(doctor) if isinstance(a, dict)]
+    except ApiError as exc:
+        alerts = []
+        st.warning(f"Алерты недоступны: {exc}")
+    ids = [int(a.get("id", -1)) for a in alerts]
+    # Alerts already present when the panel first loads are "old": the banner is only for new acute ones.
+    if "console_alerts_baseline" not in st.session_state:
+        st.session_state["console_alerts_baseline"] = max(ids, default=-1)
+    baseline = st.session_state["console_alerts_baseline"]
+    dismissed = st.session_state.get("console_alerts_dismissed", set())
+    fresh_acute = [
+        a
+        for a in alerts
+        if a.get("level") == "acute"
+        and not a.get("acked_at")
+        and baseline < int(a.get("id", -1))
+        and int(a.get("id", -1)) not in dismissed
+    ]
+    for a in fresh_acute[:1]:
+        st.markdown(
+            "<div style='background:#c62828;color:#fff;padding:18px 20px;border-radius:12px;"
+            "font-size:1.35rem;font-weight:700;margin-bottom:8px'>"
+            f"🚨 ОСТРЫЙ АЛЕРТ · {_escape(a.get('patient_name', ''))} · {_fmt_ts(a.get('ts'))}<br>"
+            f"<span style='font-size:1.05rem;font-weight:500'>{_escape(a.get('reason', ''))}</span></div>",
+            unsafe_allow_html=True,
+        )
+        c1, c2 = st.columns(2)
+        aid = int(a.get("id", -1))
+        c1.button("Взять в работу (ack)", key=f"{key}-ack{aid}", type="primary", on_click=_ack, args=(aid,))
+        c2.button("Скрыть", key=f"{key}-hide{aid}", on_click=_dismiss_banner, args=(aid,))
+
+    st.markdown("#### Пациенты")
+    try:
+        patients = [p for p in api.patients(doctor) if isinstance(p, dict)]
+    except ApiError as exc:
+        patients = []
+        st.warning(f"Список пациентов недоступен: {exc}")
+    lines = []
+    for p in patients:
+        name = str(p.get("display_name", "?"))
+        if p.get("token") == highlight_token:
+            name = f"**{name} ◀**"
+        extra = f" · {p['last_event']}" if p.get("last_event") else ""
+        lines.append(
+            f"{COLOR_EMOJI.get(p.get('status_color', ''), '⚪')} {name} · день {p.get('therapy_day', '—')}{extra}"
+        )
+    st.markdown("\n".join(f"- {line}" for line in lines) if lines else "_нет пациентов_")
+
+    st.markdown("#### Алерты")
+    if not alerts:
+        st.caption("Новых алертов нет")
+    for a in alerts[:8]:
+        done = " ✔︎" if a.get("acked_at") else ""
+        icon = LEVEL_EMOJI.get(a.get("level", ""), "⚪")
+        who = a.get("patient_name", "")
+        st.markdown(f"{icon} `{_fmt_ts(a.get('ts'))}` **{who}** — {a.get('reason', '')}{done}")
+
+
+@st.fragment(run_every=POLL_SECONDS)
+def polled_doctor_panel(highlight_token: str | None, key: str = "doc") -> None:
+    render_doctor_panel(highlight_token, key=key)
+
+
+def _escape(value: Any) -> str:
+    text = str(value)
+    for ch, rep in (("&", "&amp;"), ("<", "&lt;"), (">", "&gt;")):
+        text = text.replace(ch, rep)
+    return text
+
+
+# ------------------------------------------------------------------------------------------------ playback
+def scenario_for(persona: dict[str, Any]) -> tuple[Scenario, str]:
+    """Scenario of the persona: uploaded text (if any) wins over the repo file. Returns (scenario, source)."""
+    token = persona["token"]
+    uploaded = st.session_state.get(f"console_scenario_text:{token}")
+    if uploaded:
+        return parse_scenario(uploaded), "загруженный файл"
+    path = st.session_state.get(f"console_scenario_path:{token}") or default_scenario_path(persona)
+    if not path:
+        return Scenario(errors=["у персоны нет файла сценария"]), "—"
+    return load_scenario(path), path
+
+
+def make_runner(persona: dict[str, Any], *, delay: float) -> PlaybackRunner:
+    token = persona["token"]
+    return PlaybackRunner(get_api(), token, chat_for(token), playback_state(token), doctor=doctor_id(), delay=delay)
+
+
+def execute(logs: Any, state: PlaybackState, *, title: str) -> bool:
+    """Run a playback iterator with a live log. Returns True if it finished without a failed step."""
+    ok = True
+    state.log.clear()
+    with st.status(title, expanded=True) as status:
+        try:
+            for entry in logs:
+                line = entry.render() if isinstance(entry, StepLog) else str(entry)
+                state.log.append(line)
+                status.write(line)
+                ok = ok and entry.status != "fail"
+        except Exception as exc:  # last-resort guard: the page must stay alive
+            ok = False
+            state.log.append(f"❌ сбой проигрывания: {exc}")
+            status.write(state.log[-1])
+        status.update(label=f"{title}: {'готово' if ok else 'остановлено'}", state="complete" if ok else "error")
+    return ok

dashboard/pages/2_Демо-консоль.py

@@ -0,0 +1,34 @@
+"""Demo console (#17, ADR-004/005): web chat of the patient, demo clock and scenario playback.
+
+Everything goes through the HTTP API (`API_BASE_URL`); the page keeps working when the backend serves fixtures.
+"""
+
+import sys
+from pathlib import Path
+
+import streamlit as st
+
+# `streamlit run dashboard/app.py` already puts dashboard/ on sys.path; this keeps the page runnable on its own.
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from console import widgets as w  # noqa: E402
+from console.playback_ui import render_playback_panel  # noqa: E402
+
+st.set_page_config(page_title="Демо-консоль", page_icon="🎬", layout="wide")
+
+persona = w.select_persona()
+w.render_connect_hero()
+w.show_flash()
+st.sidebar.caption(f"API: {w.get_api().base_url}")
+
+st.title("Демо-консоль")
+st.caption("Прототип на синтетических данных. Время — виртуальные демо-часы (ADR-005).")
+w.render_clock_bar(persona, key="console-clock")
+st.divider()
+
+chat_col, play_col = st.columns([1.05, 1], gap="large")
+with chat_col:
+    st.subheader(f"💬 Чат пациента — {persona.get('display_name') or persona['token']}")
+    w.render_chat(persona["token"], height=560, key="console-chat")
+with play_col:
+    render_playback_panel(persona)

dashboard/pages/3_Сплит-экран.py

@@ -0,0 +1,58 @@
+"""Split screen for the projector (#17): patient on the left, doctor on the right, clock on top.
+
+Left: the patient web app in an iframe when `PATIENT_WEB_URL` is set (React app, e.g. http://localhost:5173),
+otherwise the console's fallback web chat. Right: compact doctor panel polled every 2 s with a red banner
+on a new acute alert.
+"""
+
+import sys
+from pathlib import Path
+
+import streamlit as st
+
+# `streamlit run dashboard/app.py` already puts dashboard/ on sys.path; this keeps the page runnable on its own.
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from console import widgets as w  # noqa: E402
+
+st.set_page_config(page_title="Сплит-экран", page_icon="🖥️", layout="wide", initial_sidebar_state="collapsed")
+st.markdown(
+    "<style>.block-container{padding-top:1.2rem;max-width:100%} "
+    "[data-testid='stMarkdownContainer'] p, [data-testid='stMarkdownContainer'] li{font-size:1.08rem}</style>",
+    unsafe_allow_html=True,
+)
+
+persona = w.select_persona()
+w.show_flash()
+token = persona["token"]
+
+top, play = st.columns([6, 1.3], vertical_alignment="center")
+with top:
+    w.render_clock_bar(persona, key="split-clock", with_reset=False)
+with play:
+    scenario, _ = w.scenario_for(persona)
+    runner = w.make_runner(persona, delay=0.8)
+    nxt = runner.next_day(scenario) if scenario.days else None
+    clicked = st.button(
+        f"▶ Сценарий: день {nxt.day}" if nxt else "Сценарий проигран",
+        disabled=nxt is None,
+        width="stretch",
+        help="Проиграть следующий день из data/synthetic/<персона>.jsonl",
+    )
+if clicked and nxt is not None:
+    w.execute(runner.play_day(scenario, nxt.day), w.playback_state(token), title=f"День {nxt.day}")
+    st.rerun()
+
+left, right = st.columns([1.1, 1], gap="large")
+with left:
+    url = w.patient_web_url()
+    if url:
+        st.markdown(f"#### 📱 Пациент — {persona.get('display_name') or token}")
+        st.iframe(url, height=760)
+        st.caption(f"Веб-приложение пациента: {url}")
+    else:
+        st.markdown(f"#### 💬 Пациент — {persona.get('display_name') or token}")
+        w.render_chat(token, height=600, key="split-chat")
+with right:
+    st.markdown("#### 🩺 Кабинет врача — д-р Смирнова Е. А.")
+    w.polled_doctor_panel(token, key="split-doc")

tests/test_dashboard_console.py

@@ -0,0 +1,419 @@
+"""Demo console (#17): scenario parser, press resolver, clock math, playback runner, API client."""
+
+import json
+import sys
+from datetime import date, datetime, timedelta, timezone
+from pathlib import Path
+from typing import Any
+
+import httpx
+import pytest
+
+# Streamlit runs pages with `dashboard/` on sys.path and they import `console`; tests mirror that.
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "dashboard"))
+
+from console.api import ApiError, ConsoleApi, personas_or_fallback  # noqa: E402
+from console.chatlog import ChatLog, normalize_reply  # noqa: E402
+from console.playback import (  # noqa: E402
+    PlaybackRunner,
+    PlaybackState,
+    hours_until,
+    parse_clock_now,
+    planned_date,
+    resolve_press,
+    target_datetime,
+)
+from console.scenario import (  # noqa: E402
+    Action,
+    default_scenario_path,
+    load_scenario,
+    parse_action,
+    parse_scenario,
+    persona_slug,
+)
+
+MSK = timezone(timedelta(hours=3))
+
+
+def _jsonl(*rows: Any) -> str:
+    return "\n".join(r if isinstance(r, str) else json.dumps(r, ensure_ascii=False) for r in rows)
+
+
+def _reply(text: str, *labels_data: tuple[str, str], kind: str = "normal") -> dict[str, Any]:
+    return {"text": text, "kind": kind, "buttons": [[{"label": lab, "data": d} for lab, d in labels_data]]}
+
+
+# ------------------------------------------------------------------------------------------------ scenario
+def test_parse_scenario_valid_sorted_and_normalized():
+    sc = parse_scenario(
+        _jsonl(
+            {"day": 2, "note": "тошнота", "actions": [{"advance_to": "8:00"}, {"press": "Принял"}]},
+            "",
+            "# comment",
+            {"day": 0, "actions": [{"start": None}, {"text": "привет"}, {"expect_color": "GREEN"}]},
+        )
+    )
+    assert sc.errors == []
+    assert sc.day_numbers == [0, 2]
+    assert sc.day(2).actions == (Action("advance_to", "08:00"), Action("press", "Принял"))
+    assert sc.day(0).actions[0] == Action("start", "")
+    assert sc.day(0).actions[2] == Action("expect_color", "green")
+    assert sc.day(2).note == "тошнота"
+
+
+def test_parse_scenario_reports_and_skips_bad_parts():
+    sc = parse_scenario(
+        _jsonl(
+            "{not json",
+            [1, 2],
+            {"day": -1, "actions": []},
+            {"day": 1, "actions": [{"advance_to": "25:00"}, {"press": "Да"}, {"fly": "x"}, {"a": 1, "b": 2}]},
+            {"day": 1, "actions": [{"text": "дубль"}]},
+            {"day": 3, "actions": "oops"},
+            {"day": 4, "actions": [{"expect_color": "purple"}, {"text": "  "}]},
+        )
+    )
+    assert sc.day_numbers == [1, 4]
+    assert sc.day(1).actions == (Action("press", "Да"),)
+    assert sc.day(4).actions == ()
+    assert len(sc.errors) == 10
+    assert any("строка 1" in e and "не JSON" in e for e in sc.errors)
+    assert any("повторяется" in e for e in sc.errors)
+    assert any("ЧЧ:ММ" in e for e in sc.errors)
+
+
+@pytest.mark.parametrize("raw", [{"press": 5}, "press", {"advance_to": "8"}, {}])
+def test_parse_action_rejects(raw):
+    with pytest.raises(ValueError):
+        parse_action(raw)
+
+
+def test_load_scenario_missing_and_present(tmp_path):
+    missing = load_scenario("nope.jsonl", root=tmp_path)
+    assert missing.days == [] and "не найден" in missing.errors[0]
+    (tmp_path / "s.jsonl").write_text(_jsonl({"day": 5, "actions": [{"text": "брошу"}]}), encoding="utf-8")
+    assert load_scenario("s.jsonl", root=tmp_path).day_numbers == [5]
+
+
+# ------------------------------------------------------------------------------------------------ press resolver
+def test_resolve_press_newest_reply_first_and_exact_beats_substring():
+    replies = [
+        _reply("вчера", ("Принял", "med:taken:1")),
+        _reply("сегодня", ("Принял позже", "med:later:2"), ("Принял", "med:taken:2")),
+    ]
+    assert resolve_press(replies, "принял") == ("med:taken:2", "Принял")
+    assert resolve_press(replies, "позже") == ("med:later:2", "Принял позже")
+
+
+def test_resolve_press_window_normalization_and_miss():
+    old = _reply("старое", ("Ещё вопрос", "q:1"))
+    filler = [_reply(f"r{i}") for i in range(10)]
+    assert resolve_press([old, *filler], "еще") is None  # outside the 10-reply window
+    assert resolve_press([old, *filler[:3]], "  ЕЩЕ  вопрос ") == ("q:1", "Ещё вопрос")
+    assert resolve_press([old], "нет такой") is None
+    assert resolve_press([old], "") is None
+
+
+# ------------------------------------------------------------------------------------------------ clock math
+def test_target_datetime_next_occurrence_without_planned_date():
+    now = datetime(2026, 10, 5, 7, 0, tzinfo=MSK)
+    assert target_datetime(now, "08:00", None) == datetime(2026, 10, 5, 8, 0, tzinfo=MSK)
+    late = datetime(2026, 10, 5, 21, 0, tzinfo=MSK)
+    assert target_datetime(late, "08:00", None) == datetime(2026, 10, 6, 8, 0, tzinfo=MSK)
+    # Just reached (virtual time ticked a bit past): stays today, no rollover.
+    just = datetime(2026, 10, 5, 8, 2, tzinfo=MSK)
+    assert target_datetime(just, "08:00", None) == datetime(2026, 10, 5, 8, 0, tzinfo=MSK)
+
+
+def test_target_datetime_with_planned_date_and_hours_until():
+    now = datetime(2026, 10, 5, 20, 30, tzinfo=MSK)
+    target = target_datetime(now, "08:00", date(2026, 10, 6))
+    assert hours_until(now, target) == pytest.approx(11.5)
+    assert hours_until(now, now - timedelta(minutes=3)) == 0.0
+    assert hours_until(now, now - timedelta(hours=2)) is None
+
+
+def test_planned_date_chains_days_and_ignores_stale_dates():
+    dates = {0: date(2026, 10, 5), 2: date(2026, 10, 7)}
+    assert planned_date(dates, 5, today=date(2026, 10, 7)) == date(2026, 10, 10)
+    assert planned_date(dates, 2, today=date(2026, 10, 7)) == date(2026, 10, 7)
+    assert planned_date({}, 3, today=date(2026, 10, 7)) is None
+    # Presenter moved the clock manually past the chained date → fall back to "next occurrence".
+    assert planned_date(dates, 3, today=date(2026, 10, 12)) is None
+
+
+def test_parse_clock_now():
+    assert parse_clock_now({"now": "2026-10-05T08:00:00+03:00"}) == datetime(2026, 10, 5, 8, 0, tzinfo=MSK)
+    with pytest.raises(ValueError):
+        parse_clock_now({})
+
+
+# ------------------------------------------------------------------------------------------------ chat log
+def test_chatlog_normalizes_and_tracks_active_keyboard():
+    assert normalize_reply({"text": None, "buttons": [[{"label": "x"}], "bad"]}) == {
+        "dir": "out",
+        "text": "",
+        "kind": "normal",
+        "buttons": [],
+    }
+    chat = ChatLog()
+    chat.add_replies([_reply("a", ("Да", "y")), {"text": "b"}])
+    chat.add_in("Да", pressed=True)
+    assert chat.last_buttons_index() == 0
+    assert [r["text"] for r in chat.recent_replies(1)] == ["b"]
+    chat.add_history([{"direction": "in", "text": "hi"}, {"direction": "out", "text": "yo", "kind": "info"}])
+    assert chat.entries[-1]["kind"] == "info" and chat.entries[-2]["dir"] == "in"
+
+
+# ------------------------------------------------------------------------------------------------ runner
+class FakeApi:
+    """Demo backend double: a virtual clock with scheduled pushes and scripted dialog answers."""
+
+    def __init__(self, now: datetime):
+        self.now = now
+        self.calls: list[tuple] = []
+        self.scheduled: list[tuple[datetime, dict[str, Any]]] = []
+        self.box: list[dict[str, Any]] = []
+        self.color = "green"
+        self.fail_on: str | None = None
+        self.answers: dict[str, list[dict[str, Any]]] = {}
+
+    def _maybe_fail(self, what: str) -> None:
+        if self.fail_on == what:
+            raise ApiError(f"{what}: HTTP 500")
+
+    def push_at(self, hhmm: str, day: date, reply: dict[str, Any]) -> None:
+        h, m = map(int, hhmm.split(":"))
+        self.scheduled.append((datetime(day.year, day.month, day.day, h, m, tzinfo=MSK), reply))
+
+    def clock(self) -> dict[str, Any]:
+        self._maybe_fail("clock")
+        return {"now": self.now.isoformat(), "virtual": True, "offset_days": 0}
+
+    def advance(self, days: float = 0, hours: float = 0) -> dict[str, Any]:
+        self._maybe_fail("advance")
+        self.calls.append(("advance", round(hours, 3)))
+        self.now += timedelta(days=days, hours=hours)
+        due = [(t, r) for t, r in self.scheduled if t <= self.now]
+        self.scheduled = [(t, r) for t, r in self.scheduled if t > self.now]
+        self.box.extend(r for _, r in due)
+        return self.clock()
+
+    def send_event(self, token, type_, *, text=None, button=None, params=None):
+        self._maybe_fail(type_)
+        self.calls.append((type_, text or button or params))
+        return self.answers.get(text or button or type_, [{"text": "ок"}])
+
+    def outbox(self, token):
+        out, self.box = self.box, []
+        return out
+
+    def patients(self, doctor):
+        self._maybe_fail("patients")
+        return [{"token": "tok-other", "status_color": "red"}, {"token": "tok-andrey", "status_color": self.color}]
+
+
+def _runner(api: FakeApi, state: PlaybackState | None = None, chat: ChatLog | None = None) -> PlaybackRunner:
+    sleeps: list[float] = []
+    r = PlaybackRunner(api, "tok-andrey", chat or ChatLog(), state or PlaybackState(), delay=0.5, sleep=sleeps.append)
+    r.sleeps = sleeps  # type: ignore[attr-defined]
+    return r
+
+
+SCENARIO = parse_scenario(
+    _jsonl(
+        {"day": 1, "note": "приём", "actions": [{"advance_to": "08:00"}, {"press": "Принял"}]},
+        {
+            "day": 2,
+            "actions": [
+                {"advance_to": "08:00"},
+                {"press": "принял"},
+                {"advance_to": "20:00"},
+                {"text": "подташнивает с утра"},
+                {"expect_color": "yellow"},
+            ],
+        },
+        {"day": 5, "actions": [{"advance_to": "20:00"}, {"text": "брошу"}]},
+    )
+)
+
+
+def test_runner_plays_day_in_order_with_dynamic_button_data():
+    api = FakeApi(datetime(2026, 10, 5, 21, 0, tzinfo=MSK))
+    api.push_at("08:00", date(2026, 10, 6), _reply("Пора принять сертралин", ("Принял", "med:taken:42")))
+    runner = _runner(api)
+    logs = list(runner.play_day(SCENARIO, 1))
+    assert api.calls == [("advance", 11.0), ("button", "med:taken:42")]
+    assert [log.status for log in logs] == ["info", "ok", "ok", "ok"]
+    assert runner.state.done == [1] and runner.state.dates[1] == date(2026, 10, 6)
+    assert runner.chat.entries[1] == {"dir": "in", "text": "Принял", "pressed": True}
+    assert runner.sleeps == [0.5]  # one pause between two steps
+
+
+def test_runner_chains_next_day_date_and_soft_color_check():
+    api = FakeApi(datetime(2026, 10, 5, 21, 0, tzinfo=MSK))
+    api.push_at("08:00", date(2026, 10, 6), _reply("доза 1", ("Принял", "med:taken:1")))
+    api.push_at("08:00", date(2026, 10, 7), _reply("доза 2", ("Принял", "med:taken:2")))
+    runner = _runner(api)
+    list(runner.play_day(SCENARIO, 1))
+    logs = list(runner.play_day(SCENARIO, 2))
+    assert api.calls[2:] == [
+        ("advance", 24.0),
+        ("button", "med:taken:2"),
+        ("advance", 12.0),
+        ("text", "подташнивает с утра"),
+    ]
+    color_log = logs[-2]
+    assert color_log.status == "warn" and "green" in color_log.message
+    assert runner.state.done == [1, 2] and runner.state.failed is None
+
+
+def test_scroll_to_plays_all_unplayed_days_fast():
+    api = FakeApi(datetime(2026, 10, 5, 7, 0, tzinfo=MSK))
+    api.color = "yellow"
+    for d in (5, 6):
+        api.push_at("08:00", date(2026, 10, d), _reply("доза", ("Принял", f"med:taken:{d}")))
+    runner = _runner(api)
+    runner.delay = 0
+    logs = list(runner.scroll_to(SCENARIO, 5))
+    assert runner.state.done == [1, 2, 5]
+    # Day 1 lands on today 08:00 (nothing played yet), day 5 is chained: 5 − 2 = 3 days after day 2.
+    assert runner.state.dates == {1: date(2026, 10, 5), 2: date(2026, 10, 6), 5: date(2026, 10, 9)}
+    assert all(log.status != "fail" for log in logs)
+    assert runner.sleeps == []
+    assert list(runner.scroll_to(SCENARIO, 5))[0].status == "info"  # nothing left to scroll
+
+
+def test_missing_button_pauses_and_resume_can_skip_or_retry():
+    api = FakeApi(datetime(2026, 10, 5, 21, 0, tzinfo=MSK))  # no reminder scheduled → button never shows up
+    runner = _runner(api)
+    logs = list(runner.play_day(SCENARIO, 1))
+    assert logs[-1].status == "fail" and "не найдена" in logs[-1].message
+    assert runner.state.failed is not None and (runner.state.failed.day, runner.state.failed.step) == (1, 1)
+    assert runner.state.done == []
+    assert runner.sleeps == [0.5] + [runner.retry_wait] * runner.press_retries  # step pause, then retries
+
+    api.box.append(_reply("доза", ("Принял", "med:taken:7")))
+    retry = list(runner.resume(SCENARIO, skip=False))
+    assert retry[-1].status == "ok" and ("button", "med:taken:7") in api.calls
+    assert runner.state.failed is None and runner.state.done == [1]
+
+
+def test_resume_skip_continues_until_target_day():
+    api = FakeApi(datetime(2026, 10, 5, 21, 0, tzinfo=MSK))
+    runner = _runner(api)
+    runner.delay = 0
+    list(runner.scroll_to(SCENARIO, 2))  # fails on day 1 «Принял»
+    assert runner.state.failed.until == 2
+    list(runner.resume(SCENARIO, skip=True))  # skip the press, play day 1 rest and day 2 → fails on day 2 press
+    assert runner.state.done == [1]
+    assert (runner.state.failed.day, runner.state.failed.step) == (2, 1)
+
+
+def test_api_errors_never_escape_the_runner():
+    api = FakeApi(datetime(2026, 10, 5, 19, 0, tzinfo=MSK))
+    api.fail_on = "text"
+    runner = _runner(api)
+    logs = list(runner.play_day(SCENARIO, 5))
+    assert logs[-1].status == "fail" and "HTTP 500" in logs[-1].message
+    assert runner.state.failed is not None
+
+    api.fail_on = "clock"
+    logs = list(_runner(api).play_day(SCENARIO, 5))
+    assert logs[-1].status == "fail" and "часы" in logs[-1].message
+
+    api.fail_on = "patients"
+    color_only = parse_scenario(_jsonl({"day": 0, "actions": [{"expect_color": "green"}]}))
+    logs = list(_runner(api).play_day(color_only, 0))
+    assert logs[1].status == "warn" and logs[-1].status == "ok"
+
+
+def test_past_target_is_skipped_with_warning():
+    api = FakeApi(datetime(2026, 10, 6, 21, 0, tzinfo=MSK))
+    state = PlaybackState(dates={5: date(2026, 10, 6)})
+    logs = list(_runner(api, state).play_day(SCENARIO, 5))  # replaying day 5 after 20:00 the same evening
+    assert logs[1].status == "warn" and "уже прошло" in logs[1].message
+    assert ("text", "брошу") in api.calls and not any(c[0] == "advance" for c in api.calls)
+
+
+def test_play_unknown_day_is_info():
+    logs = list(_runner(FakeApi(datetime(2026, 10, 5, 8, 0, tzinfo=MSK))).play_day(SCENARIO, 42))
+    assert [log.status for log in logs] == ["info"]
+
+
+def test_start_action_sends_code():
+    api = FakeApi(datetime(2026, 10, 5, 8, 0, tzinfo=MSK))
+    sc = parse_scenario(_jsonl({"day": 0, "actions": [{"start": "DEMO-0001"}, {"start": ""}]}))
+    list(_runner(api).play_day(sc, 0))
+    assert api.calls == [("start", {"code": "DEMO-0001"}), ("start", {})]
+
+
+# ------------------------------------------------------------------------------------------------ http client
+def _client(handler) -> ConsoleApi:
+    return ConsoleApi("http://api", client=httpx.Client(transport=httpx.MockTransport(handler)))
+
+
+def test_console_api_paths_and_bodies():
+    seen: list[tuple[str, str, Any]] = []
+
+    def handler(req: httpx.Request) -> httpx.Response:
+        body = json.loads(req.content) if req.content else None
+        seen.append((req.method, req.url.path, body))
+        if req.url.path.endswith("/events"):
+            return httpx.Response(200, json={"replies": [{"text": "hi"}]})
+        if req.url.path.endswith("/outbox"):
+            return httpx.Response(200, json={"replies": []})
+        return httpx.Response(200, json={"now": "2026-10-05T08:00:00+03:00"})
+
+    api = _client(handler)
+    assert api.send_event("tok-andrey", "button", button="med:taken:1") == [{"text": "hi"}]
+    assert api.outbox("tok-andrey") == []
+    api.advance(hours=1.5)
+    assert seen == [
+        ("POST", "/api/v1/dialog/tok-andrey/events", {"type": "button", "params": {}, "button": "med:taken:1"}),
+        ("GET", "/api/v1/dialog/tok-andrey/outbox", None),
+        ("POST", "/api/v1/demo/clock/advance", {"days": 0, "hours": 1.5}),
+    ]
+
+
+def test_console_api_errors_become_api_error_and_personas_fall_back(monkeypatch):
+    def down(req: httpx.Request) -> httpx.Response:
+        raise httpx.ConnectError("refused")
+
+    api = _client(down)
+    with pytest.raises(ApiError, match="недоступен"):
+        api.clock()
+    assert api.health() is False
+    monkeypatch.setenv("DEMO_PERSONA_TOKEN", "tok-andrey")
+    personas, error = personas_or_fallback(api)
+    assert personas[0]["token"] == "tok-andrey" and error
+
+    api = _client(lambda req: httpx.Response(500))
+    with pytest.raises(ApiError, match="HTTP 500"):
+        api.outbox("t")
+
+    ok = _client(lambda req: httpx.Response(200, json=[{"token": "tok-andrey", "display_name": "Андрей С."}]))
+    personas, error = personas_or_fallback(ok)
+    assert error is None and personas[0]["display_name"] == "Андрей С."
+
+
+def test_personas_fall_back_to_doctor_patients_with_hero_first():
+    def handler(req: httpx.Request) -> httpx.Response:
+        if req.url.path.endswith("/demo/personas"):
+            return httpx.Response(200, json=[])
+        return httpx.Response(
+            200, json=[{"token": "h1", "display_name": "Сергей П."}, {"token": "h2", "display_name": "Андрей С."}]
+        )
+
+    personas, error = personas_or_fallback(_client(handler))
+    assert [p["token"] for p in personas] == ["h2", "h1"] and "пациенты врача" in error
+
+
+def test_scenario_path_guessed_from_name_or_token(tmp_path):
+    (tmp_path / "data" / "synthetic").mkdir(parents=True)
+    (tmp_path / "data" / "synthetic" / "andrey.jsonl").write_text("", encoding="utf-8")
+    assert persona_slug("Андрей С.") == "andrey" and persona_slug("Наталья Ф.") == "natalya"
+    assert default_scenario_path({"token": "h", "display_name": "Андрей С."}, tmp_path) == "data/synthetic/andrey.jsonl"
+    assert default_scenario_path({"token": "tok-andrey"}, tmp_path) == "data/synthetic/andrey.jsonl"
+    assert default_scenario_path({"token": "x", "display_name": "Ольга Н."}, tmp_path) is None
+    assert default_scenario_path({"token": "x", "scenario": "s.jsonl"}, tmp_path) == "s.jsonl"

tests/test_dashboard_console_pages.py

@@ -0,0 +1,64 @@
+"""Demo console pages (#17) render without crashing even when the API is down (robustness of the live demo)."""
+
+from pathlib import Path
+
+import pytest
+from streamlit.testing.v1 import AppTest
+
+PAGES = Path(__file__).resolve().parents[1] / "dashboard" / "pages"
+
+
+@pytest.fixture
+def api_down(monkeypatch):
+    monkeypatch.setenv("API_BASE_URL", "http://127.0.0.1:9")  # nothing listens on the discard port
+    monkeypatch.setenv("DEMO_PERSONA_TOKEN", "tok-andrey")
+    monkeypatch.delenv("PATIENT_WEB_URL", raising=False)
+
+
+def _page(name: str) -> AppTest:
+    return AppTest.from_file(str(PAGES / name), default_timeout=30)
+
+
+def test_console_page_survives_api_down_and_button_clicks(api_down, tmp_path):
+    scenario = tmp_path / "s.jsonl"
+    scenario.write_text('{"day": 1, "note": "утро", "actions": [{"advance_to": "08:00"}]}\n', encoding="utf-8")
+    at = _page("2_Демо-консоль.py")
+    at.session_state["console_scenario_path:tok-andrey"] = str(scenario)
+    at.run()
+    assert not at.exception
+    assert any("Часы недоступны" in w.value for w in at.warning)
+    labels = [b.label for b in at.button]
+    assert {"+1 час", "Следующий день", "🚀 Начать", "▶ Проиграть день 1"} <= set(labels)
+
+    at.button(key="console-clock-d1").click().run()  # «Следующий день» with the API down
+    assert not at.exception
+    at.button(key="console-chat-start").click().run()
+    assert not at.exception
+    next(b for b in at.button if b.label == "▶ Проиграть день 1").click().run()
+    assert not at.exception
+    state = at.session_state["console_playback:tok-andrey"]
+    assert state.failed is not None and state.done == []  # playback paused, page alive
+    at.run()
+    assert not at.exception
+    assert any("Повторить шаг" in b.label for b in at.button), [b.label for b in at.button]
+
+
+def test_console_page_without_scenario_file(api_down):
+    at = _page("2_Демо-консоль.py")
+    at.session_state["console_scenario_path:tok-andrey"] = "data/synthetic/__missing__.jsonl"
+    at.run()
+    assert not at.exception
+    assert any("Сценарий не загружен" in i.value for i in at.info)
+
+
+def test_split_screen_survives_api_down_and_uses_iframe_when_configured(api_down, monkeypatch):
+    at = _page("3_Сплит-экран.py")
+    at.run()
+    assert not at.exception
+    assert any("Алерты недоступны" in w.value for w in at.warning)
+
+    monkeypatch.setenv("PATIENT_WEB_URL", "http://localhost:5173")
+    at = _page("3_Сплит-экран.py")
+    at.run()
+    assert not at.exception
+    assert any("http://localhost:5173" in c.value for c in at.caption)