feat: live agent answers (SSE streaming) + «Чат» tab on phones

- POST /dialog/{token}/events/stream: Server-Sent Events with the agent's steps («Смотрю вашу карту ЕМИАС…»,
  «Ищу в клинических рекомендациях…»), the answer as the model writes it (OpenRouter stream=true), then the
  final replies (same as /events, incl. the forbidden-output filter and sources). X-Accel-Buffering: no.
- Web: free-text messages use the stream — a live bubble with steps, typing dots and a cursor; falls back to
  the plain request if the stream breaks.
- Phones: the bottom tab bar gets «Чат» (it had no way back to the chat).

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

app/api/routes.py

@@ -8,9 +8,14 @@ disclosed — the dashboard polls every 2–3 s), `ack_alert` (in `repo.ack_aler
 `app/api/fixtures.py` is kept for tests and as the demo cabinet reference (doctor id).
 """
 
+import asyncio
+import json
+import logging
 import secrets
+from typing import Any
 
 from fastapi import APIRouter, HTTPException
+from fastapi.responses import StreamingResponse
 from sqlalchemy import select
 
 from app.api import demo as console
@@ -42,10 +47,12 @@ from app.db.identity import Patient
 from app.db.session import get_clinical_session, get_identity_session
 from app.dialog.contracts import Event, get_engine
 from app.integrations.base import integrations
+from app.llm.agent import stream as agent_stream
 from app.scheduler import jobs as scheduler
 from app.services.anonymous_care import display_name as anonymous_name
 from app.services.monitoring import refresh_status
 
+log = logging.getLogger(__name__)
 router = APIRouter(prefix="/api/v1")
 router.include_router(max_notifications.router)
 router.include_router(questionnaire_router)
@@ -219,6 +226,49 @@ async def dialog_event(token: str, body: DialogIn) -> DialogOut:
     return DialogOut(replies=await get_engine().handle(token, body))
 
 
+@router.post("/dialog/{token}/events/stream", tags=["dialog"])
+async def dialog_event_stream(token: str, body: DialogIn) -> StreamingResponse:
+    """Same as /events, as Server-Sent Events: `status` (agent steps), `delta` (answer as it is written),
+    `reset`, then `done` with the final replies (identical to /events). `error` if the engine failed."""
+    await _require_patient(token)
+    queue: asyncio.Queue[tuple[str, Any]] = asyncio.Queue()
+    marker = agent_stream.STREAM.set(queue)  # copied into the task's context below
+    try:
+
+        async def run() -> None:
+            try:
+                replies = await get_engine().handle(token, body)
+                queue.put_nowait(("done", [r.model_dump() for r in replies]))
+            except Exception:  # noqa: BLE001 — reported to the client, which falls back to /events
+                log.exception("dialog stream failed")
+                queue.put_nowait(("error", None))
+
+        task = asyncio.create_task(run())
+    finally:
+        agent_stream.STREAM.reset(marker)
+
+    async def events():
+        try:
+            while True:
+                try:
+                    kind, payload = await asyncio.wait_for(queue.get(), timeout=15)
+                except TimeoutError:
+                    yield ": keepalive\n\n"
+                    continue
+                yield f"event: {kind}\ndata: {json.dumps(payload, ensure_ascii=False)}\n\n"
+                if kind in ("done", "error"):
+                    return
+        finally:
+            if not task.done():
+                await task
+
+    return StreamingResponse(
+        events(),
+        media_type="text/event-stream",
+        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},  # nginx: do not buffer SSE
+    )
+
+
 @router.get("/dialog/{token}/outbox", response_model=DialogOut, tags=["dialog"])
 async def dialog_outbox(token: str) -> DialogOut:
     """System-initiated replies (reminders, check-ins) waiting for the web console."""

app/llm/agent/__init__.py

@@ -21,6 +21,7 @@ from pydantic import BaseModel
 
 from app.config import Settings, get_settings
 from app.core.types import ScrubbedText
+from app.llm.agent import stream
 from app.llm.agent.prompts import SYSTEM_PROMPT
 from app.llm.agent.providers import (
     GigaChatToolProvider,
@@ -127,6 +128,7 @@ async def answer_chat(
         return None
     if is_small_talk(question):
         return AgentAnswer(text=SMALL_TALK_REPLY, sources=[], tools=[], provider="small_talk")
+    stream.emit("status", "Думаю над вопросом…")
     ctx = ToolContext(token=token)
     messages: list[dict[str, Any]] = [
         {"role": "system", "content": SYSTEM_PROMPT + await _patient_brief(ctx)},
@@ -159,6 +161,7 @@ async def answer_chat(
                     }
                 )
                 for c in turn.tool_calls[:MAX_TOOL_CALLS_PER_TURN]:
+                    stream.emit("status", stream.TOOL_STATUS.get(c.name, "Проверяю данные…"))
                     result = await run_tool(ctx, c.name, c.arguments)
                     messages.append({"role": "tool", "tool_call_id": c.id, "name": c.name, "content": result})
             log.warning("chat agent: step limit reached tools=%s", ",".join(ctx.used))

app/llm/agent/providers.py

@@ -16,6 +16,7 @@ import httpx
 from pydantic import BaseModel, Field
 
 from app.config import Settings
+from app.llm.agent import stream
 
 log = logging.getLogger(__name__)
 
@@ -53,6 +54,8 @@ class OpenRouterProvider:
         self._client = client
 
     async def chat(self, messages: list[dict[str, Any]], tools: list[dict[str, Any]]) -> AssistantTurn:
+        if stream.active():
+            return await self._chat_stream(messages, tools)
         body: dict[str, Any] = {"model": self._model, "messages": messages, "tools": tools, "temperature": 0.1}
         body["max_tokens"] = 900
         if not any(m["role"] == "tool" for m in messages):
@@ -174,6 +177,78 @@ class GigaChatToolProvider:
         return AssistantTurn(content=msg.content or None, tool_calls=calls)
 
 
+# ------------------------------------------------------------------------------------------------ streaming
+
+
+async def _iter_sse(resp: httpx.Response):
+    async for line in resp.aiter_lines():
+        if not line.startswith("data:"):
+            continue  # comments («: OPENROUTER PROCESSING») and blank lines
+        data = line[5:].strip()
+        if data == "[DONE]":
+            return
+        try:
+            yield json.loads(data)
+        except json.JSONDecodeError:
+            continue
+
+
+async def _openrouter_stream(self: "OpenRouterProvider", messages: list[dict[str, Any]], tools: list[dict[str, Any]]):
+    """Same request with `stream: true`: content deltas go to the live channel, tool calls are assembled."""
+    body: dict[str, Any] = {
+        "model": self._model,
+        "messages": messages,
+        "tools": tools,
+        "temperature": 0.1,
+        "max_tokens": 900,
+        "stream": True,
+    }
+    if not any(m["role"] == "tool" for m in messages):
+        body["tool_choice"] = "required"
+    headers = {"Authorization": f"Bearer {self._key}", "X-Title": "PTSRaiAgent prototype (synthetic data)"}
+    content: list[str] = []
+    calls: dict[int, dict[str, Any]] = {}
+    try:
+        client = self._client or httpx.AsyncClient(timeout=self._timeout, proxy=self._proxy)
+        try:
+            async with client.stream("POST", self._url, json=body, headers=headers, timeout=self._timeout) as resp:
+                if resp.status_code != 200:
+                    await resp.aread()
+                    raise ProviderError(f"http {resp.status_code}")
+                async for chunk in _iter_sse(resp):
+                    choice = (chunk.get("choices") or [{}])[0]
+                    delta = choice.get("delta") or {}
+                    if delta.get("content"):
+                        content.append(delta["content"])
+                        stream.emit("delta", delta["content"])
+                    for tc in delta.get("tool_calls") or []:
+                        slot = calls.setdefault(tc.get("index", 0), {"id": "", "name": "", "args": ""})
+                        slot["id"] = tc.get("id") or slot["id"]
+                        fn = tc.get("function") or {}
+                        slot["name"] += fn.get("name") or ""
+                        slot["args"] += fn.get("arguments") or ""
+        finally:
+            if self._client is None:
+                await client.aclose()
+    except httpx.HTTPError as exc:
+        raise ProviderError(type(exc).__name__) from exc
+    out = []
+    for i in sorted(calls):
+        c = calls[i]
+        try:
+            args = json.loads(c["args"] or "{}")
+        except json.JSONDecodeError:
+            args = {}
+        out.append(ToolCall(id=c["id"] or c["name"] or f"call{i}", name=c["name"], arguments=args))
+    if out and content:
+        stream.emit("reset", None)  # a preface before tool calls is not the answer
+    log.info("openrouter stream ok calls=%d chars=%d", len(out), sum(map(len, content)))
+    return AssistantTurn(content="".join(content) or None, tool_calls=out)
+
+
+OpenRouterProvider._chat_stream = _openrouter_stream  # type: ignore[attr-defined]
+
+
 # ------------------------------------------------------------------------------------------------ offline mock
 
 _DRUG_WORDS = re.compile(

app/llm/agent/stream.py

@@ -0,0 +1,32 @@
+"""Live progress of the chat agent for streaming channels (web SSE).
+
+The SSE endpoint sets a queue in the context; the agent pushes events while it works:
+    status — a tool step in plain Russian («Смотрю вашу карту ЕМИАС…»)
+    delta  — a chunk of the answer as the model writes it
+    reset  — the streamed text so far was a preface before tool calls: clear it
+Channels without a queue (Telegram, tests) are unaffected. Deltas are the raw model text: the final reply
+(forbidden-output filter, sources footer) always comes in the closing `done` event and replaces it.
+"""
+
+import asyncio
+from contextvars import ContextVar
+from typing import Any
+
+STREAM: ContextVar[asyncio.Queue[tuple[str, Any]] | None] = ContextVar("agent_stream", default=None)
+
+TOOL_STATUS = {
+    "get_patient_record": "Смотрю вашу карту ЕМИАС…",
+    "check_drug_for_patient": "Проверяю препарат по вашей карте: болезни, непереносимости, другие лекарства…",
+    "get_drug_info": "Открываю инструкцию к препарату (ГРЛС)…",
+    "search_guidelines": "Ищу в клинических рекомендациях Минздрава…",
+}
+
+
+def active() -> bool:
+    return STREAM.get() is not None
+
+
+def emit(kind: str, payload: Any) -> None:
+    q = STREAM.get()
+    if q is not None:
+        q.put_nowait((kind, payload))

tests/test_agent_stream.py

@@ -0,0 +1,83 @@
+"""Live agent output: OpenRouter stream parsing and the SSE endpoint (status → delta → done)."""
+
+import asyncio
+import json
+
+import httpx
+import pytest
+from httpx import ASGITransport, AsyncClient
+
+from app.config import Settings
+from app.db import session
+from app.db.bootstrap import init_db
+from app.llm import agent
+from app.llm.agent import stream
+from app.llm.agent.providers import OpenRouterProvider
+from app.main import app
+
+MEM = "sqlite+aiosqlite:///:memory:"
+
+
+def _sse(*chunks: dict) -> bytes:
+    lines = [": OPENROUTER PROCESSING", ""]
+    lines += [f"data: {json.dumps(c, ensure_ascii=False)}" for c in chunks] + ["data: [DONE]", ""]
+    return "\n\n".join(lines).encode()
+
+
+async def test_openrouter_stream_deltas_and_tool_calls():
+    body = _sse(
+        {"choices": [{"delta": {"content": "Сейчас "}}]},
+        {
+            "choices": [
+                {
+                    "delta": {
+                        "tool_calls": [{"index": 0, "id": "c1", "function": {"name": "get_patient_", "arguments": ""}}]
+                    }
+                }
+            ]
+        },
+        {"choices": [{"delta": {"tool_calls": [{"index": 0, "function": {"name": "record", "arguments": "{}"}}]}}]},
+    )
+    client = httpx.AsyncClient(transport=httpx.MockTransport(lambda r: httpx.Response(200, content=body)))
+    provider = OpenRouterProvider(Settings(_env_file=None, openrouter_api_key="k"), client=client)
+    q: asyncio.Queue = asyncio.Queue()
+    token = stream.STREAM.set(q)
+    try:
+        turn = await provider.chat([{"role": "user", "content": "x"}], [])
+    finally:
+        stream.STREAM.reset(token)
+    assert [c.name for c in turn.tool_calls] == ["get_patient_record"]
+    events = [q.get_nowait() for _ in range(q.qsize())]
+    assert ("delta", "Сейчас ") in events and ("reset", None) in events
+
+
+@pytest.fixture
+async def client():
+    await session.configure(identity_url=MEM, clinical_url=MEM)
+    await init_db()
+    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://t") as c:
+        yield c
+    await session.configure()
+
+
+async def test_sse_endpoint_streams_steps_then_final(client, monkeypatch):
+    async def fake(token, text, **kw):
+        stream.emit("status", stream.TOOL_STATUS["get_patient_record"])
+        stream.emit("delta", "Ответ ")
+        stream.emit("delta", "агента")
+        return agent.AgentAnswer(text="Ответ агента", sources=[], tools=["get_patient_record"], provider="fake")
+
+    monkeypatch.setattr(agent, "answer_chat", fake)
+    token = (await client.post("/api/v1/auth/demo/fresh", json={})).json()["token"]
+    kinds = []
+    async with client.stream(
+        "POST", f"/api/v1/dialog/{token}/events/stream", json={"type": "text", "text": "Можно ибупрофен?"}
+    ) as resp:
+        assert resp.headers["content-type"].startswith("text/event-stream")
+        async for line in resp.aiter_lines():
+            if line.startswith("event:"):
+                kinds.append(line[6:].strip())
+            if line.startswith("data:") and kinds and kinds[-1] == "done":
+                final = json.loads(line[5:])
+    assert kinds[0] == "status" and "delta" in kinds and kinds[-1] == "done"
+    assert any("Ответ агента" in r["text"] for r in final)

web/src/api/client.ts

@@ -59,3 +59,53 @@ export const api = {
   outbox: async (token: string) =>
     (await request<{ replies: Reply[] }>(`/dialog/${encodeURIComponent(token)}/outbox`)).replies,
 };
+
+export type StreamEvent =
+  | { kind: "status"; text: string }
+  | { kind: "delta"; text: string }
+  | { kind: "reset" }
+  | { kind: "done"; replies: Reply[] };
+
+/** POST /dialog/{token}/events/stream (Server-Sent Events): agent steps and the answer as it is written,
+ *  then the final replies. Resolves with the final replies; throws if the stream breaks (caller falls back). */
+export async function sendEventStream(
+  token: string,
+  event: DialogEvent,
+  onEvent: (e: StreamEvent) => void,
+): Promise<Reply[]> {
+  const res = await fetch(`${BASE}/dialog/${token}/events/stream`, {
+    method: "POST",
+    headers: { "Content-Type": "application/json", Accept: "text/event-stream" },
+    body: JSON.stringify(event),
+  });
+  if (!res.ok || !res.body) throw new ApiError(res.status, res.statusText);
+  const reader = res.body.getReader();
+  const decoder = new TextDecoder();
+  let buf = "";
+  for (;;) {
+    const { value, done } = await reader.read();
+    if (done) break;
+    buf += decoder.decode(value, { stream: true });
+    let cut: number;
+    while ((cut = buf.indexOf("\n\n")) >= 0) {
+      const block = buf.slice(0, cut);
+      buf = buf.slice(cut + 2);
+      let kind = "message";
+      let data = "";
+      for (const line of block.split("\n")) {
+        if (line.startsWith("event:")) kind = line.slice(6).trim();
+        else if (line.startsWith("data:")) data += line.slice(5).trim();
+      }
+      if (!data) continue;
+      const payload = JSON.parse(data);
+      if (kind === "status") onEvent({ kind, text: String(payload) });
+      else if (kind === "delta") onEvent({ kind, text: String(payload) });
+      else if (kind === "reset") onEvent({ kind: "reset" });
+      else if (kind === "done") {
+        onEvent({ kind: "done", replies: payload as Reply[] });
+        return payload as Reply[];
+      } else if (kind === "error") throw new ApiError(500, "stream error");
+    }
+  }
+  throw new ApiError(0, "stream closed");
+}

web/src/components/AppShell.tsx

@@ -16,6 +16,7 @@ import { Icon, LogoMark, type IconName } from "./Icon";
 
 const TABS: { to: string; label: string; icon: IconName }[] = [
   { to: "/today", label: ru.nav.today, icon: "home" },
+  { to: "/messages", label: ru.nav.chat, icon: "message" },
   { to: "/meds", label: ru.nav.meds, icon: "pill" },
   { to: "/doctor", label: ru.nav.doctor, icon: "doctor" },
   { to: "/profile", label: ru.nav.profile, icon: "user" },

web/src/content/ru.ts

@@ -16,6 +16,7 @@ export const ru = {
     doctor: "Врач",
     profile: "Профиль",
     messages: "Сообщения",
+    chat: "Чат",
     signIn: "Вход",
     help: "Помощь",
   },
@@ -251,6 +252,7 @@ export const ru = {
     noticeTitle: "Опрос приостановлен",
   },
   messages: {
+    typing: "Помощник печатает",
     title: "Сообщения",
     inputLabel: "Сообщение",
     inputPlaceholder: "Напишите сообщение",

web/src/dialog/DialogContext.tsx

@@ -1,6 +1,6 @@
 import { createContext, useCallback, useContext, useEffect, useMemo, useRef, useState, type ReactNode } from "react";
 
-import { api } from "../api/client";
+import { api, sendEventStream } from "../api/client";
 import type { DialogEvent, Reply } from "../api/types";
 import { useSession } from "../session/SessionContext";
 import { sessionStore, type ChatEntry, type DialogSnapshot } from "../session/store";
@@ -17,8 +17,15 @@ let seq = 0;
 const newId = () => `${Date.now().toString(36)}-${(seq++).toString(36)}`;
 const asEntries = (replies: Reply[]): ChatEntry[] => replies.map((reply) => ({ id: newId(), from: "assistant", reply }));
 
+/** The agent at work: its steps and the answer as it is being written (cleared when the final reply lands). */
+export interface LiveReply {
+  status: string[];
+  text: string;
+}
+
 interface DialogApi extends DialogSnapshot {
   busy: boolean;
+  live: LiveReply | null;
   error: boolean;
   started: boolean;
   /** Sends an event; `echo` is what the patient «said» (button label / text) for the history. */
@@ -36,6 +43,7 @@ export function DialogProvider({ children }: { children: ReactNode }) {
   const [owner, setOwner] = useState(token);
   const [snap, setSnap] = useState<DialogSnapshot>(() => (token ? sessionStore.loadDialog(token) : EMPTY));
   const [busy, setBusy] = useState(false);
+  const [live, setLive] = useState<LiveReply | null>(null);
   const [error, setError] = useState(false);
   const tokenRef = useRef(token);
   tokenRef.current = token;
@@ -58,7 +66,35 @@ export function DialogProvider({ children }: { children: ReactNode }) {
       setBusy(true);
       setError(false);
       try {
-        const replies = await api.sendEvent(t, event);
+        let replies: Reply[];
+        if (event.type === "text") {
+          // Free text may go to the agent: stream its steps and the answer so the viewer sees it working.
+          const said = echo;
+          if (said) {
+            // Show the patient's message right away, while the agent is still working.
+            setSnap((prev) => ({
+              ...prev,
+              entries: [...prev.entries, { id: newId(), from: "patient" as const, text: said }],
+            }));
+            echo = undefined;
+          }
+          setLive({ status: [], text: "" });
+          try {
+            replies = await sendEventStream(t, event, (e) =>
+              setLive((prev) => {
+                const cur = prev ?? { status: [], text: "" };
+                if (e.kind === "status") return { ...cur, status: [...cur.status, e.text] };
+                if (e.kind === "delta") return { ...cur, text: cur.text + e.text };
+                if (e.kind === "reset") return { ...cur, text: "" };
+                return cur;
+              }),
+            );
+          } catch {
+            replies = await api.sendEvent(t, event); // proxies without SSE support: plain request
+          }
+        } else {
+          replies = await api.sendEvent(t, event);
+        }
         setSnap((prev) => ({
           entries: [
             ...prev.entries,
@@ -73,6 +109,7 @@ export function DialogProvider({ children }: { children: ReactNode }) {
         setError(true);
         return false;
       } finally {
+        setLive(null);
         setBusy(false);
       }
     },
@@ -102,8 +139,8 @@ export function DialogProvider({ children }: { children: ReactNode }) {
   }, [token]);
 
   const value = useMemo<DialogApi>(
-    () => ({ ...snap, busy, error, started: snap.entries.length > 0, send, start, markRead }),
+    () => ({ ...snap, busy, live, error, started: snap.entries.length > 0, send, start, markRead }),
-    [snap, busy, error, send, start, markRead],
+    [snap, busy, live, error, send, start, markRead],
   );
   return <Ctx.Provider value={value}>{children}</Ctx.Provider>;
 }

web/src/screens/Messages.tsx

@@ -18,7 +18,7 @@ export function Messages() {
   useEffect(() => {
     dialog.markRead();
     endRef.current?.scrollIntoView({ block: "end" });
-  }, [dialog.entries.length]);
+  }, [dialog.entries.length, dialog.live?.text, dialog.live?.status.length]);
 
   const send = async (e?: FormEvent) => {
     e?.preventDefault();
@@ -56,6 +56,32 @@ export function Messages() {
               </li>
             ),
           )}
+          {dialog.live && (
+            <li className="bubble bubble--bot bubble--live" aria-busy="true">
+              <span className="bubble__avatar">
+                <Icon name="smile" size={20} />
+              </span>
+              <div className="live">
+                {dialog.live.status.map((st, i) => (
+                  <p key={i} className={`live__step${i === dialog.live!.status.length - 1 && !dialog.live!.text ? " is-active" : ""}`}>
+                    {st}
+                  </p>
+                ))}
+                {dialog.live.text ? (
+                  <p className="live__text">
+                    {dialog.live.text}
+                    <span className="live__cursor" aria-hidden="true" />
+                  </p>
+                ) : (
+                  <p className="live__dots" aria-label={t.typing}>
+                    <span />
+                    <span />
+                    <span />
+                  </p>
+                )}
+              </div>
+            </li>
+          )}
         </ol>
         {quick.length > 0 && (
           <div className="quickreplies">

web/src/theme/global.css

@@ -103,7 +103,7 @@ button:disabled {
   inset: auto 0 0 0;
   z-index: 20;
   display: grid;
-  grid-template-columns: repeat(4, 1fr);
+  grid-template-columns: repeat(5, 1fr);
   height: calc(var(--nav-h) + var(--safe-bottom));
   padding-bottom: var(--safe-bottom);
   background: var(--c-surface);
@@ -2937,3 +2937,19 @@ button.row-card:hover:not(:disabled) {
     animation: none;
   }
 }
+
+
+/* Live agent reply: steps + the answer as it is written (SSE) */
+.bubble--live .live { display: grid; gap: 4px; }
+.live__step { margin: 0; font-size: 13px; color: var(--muted, #6b7280); }
+.live__step.is-active::after { content: ""; display: inline-block; width: 1em; animation: live-ellipsis 1.2s steps(4) infinite; overflow: hidden; vertical-align: bottom; }
+.live__text { margin: 2px 0 0; white-space: pre-wrap; }
+.live__cursor { display: inline-block; width: 7px; height: 1em; margin-left: 2px; background: currentColor; vertical-align: text-bottom; animation: live-blink 1s steps(2) infinite; }
+.live__dots { display: inline-flex; gap: 4px; margin: 4px 0 0; }
+.live__dots span { width: 6px; height: 6px; border-radius: 50%; background: var(--muted, #9ca3af); animation: live-dot 1.2s infinite ease-in-out; }
+.live__dots span:nth-child(2) { animation-delay: .15s; }
+.live__dots span:nth-child(3) { animation-delay: .3s; }
+@keyframes live-blink { 50% { opacity: 0; } }
+@keyframes live-dot { 0%, 80%, 100% { opacity: .3; transform: translateY(0); } 40% { opacity: 1; transform: translateY(-3px); } }
+@keyframes live-ellipsis { from { width: 0; } to { width: 1em; } }
+.tabbar__item span { font-size: 11px; }