-
`;
messagesEl.appendChild(wrap);
- wrap.scrollIntoView({ behavior: "smooth", block: "end" });
+ wrap.scrollIntoView({ behavior: SCROLL, block: "end" });
}
function removeTyping() {
@@ -164,12 +207,68 @@ async function loadHealth() {
}
}
-/* ---------- composer ---------- */
-function setBusy(busy) {
- sendBtn.disabled = busy;
- sendBtn.querySelector(".spinner").hidden = !busy;
- sendLabel.textContent = busy ? "Thinking…" : "Send";
- sendStatus.textContent = busy ? "Brain of Reese is working" : "";
+/* ---------- chat feedback state machine (PLAN §7.4) ----------
+ *
+ * Timers belong to the state machine, not to the turn handler: every
+ * transition stops/clears them, which is what makes a stuck button
+ * impossible.
+ */
+let uiState = UI_STATE.idle;
+let thinkingClock = 0; // setInterval id — elapsed-seconds hint
+let thinkingStart = 0; // Date.now() when "thinking" began
+let turnTimeout = 0; // setTimeout id — 120s pre-token guard
+
+function stopThinkingClock() {
+ if (thinkingClock) {
+ clearInterval(thinkingClock);
+ thinkingClock = 0;
+ }
+}
+
+function startThinkingClock() {
+ thinkingStart = Date.now();
+ thinkingClock = setInterval(() => {
+ const secs = Math.round((Date.now() - thinkingStart) / 1000);
+ if (secs < 10) return; // hint only after 10s of pre-token silence
+ const bubble = document.querySelector("#typing-indicator .bubble");
+ if (bubble) {
+ bubble.setAttribute("aria-label", `Brain of Reese is still thinking (${secs}s)`);
+ }
+ }, 1000);
+}
+
+function armTurnTimeout(onTimeout) {
+ clearTurnTimeout();
+ turnTimeout = setTimeout(onTimeout, TURN_TIMEOUT_MS);
+}
+
+function clearTurnTimeout() {
+ if (turnTimeout) {
+ clearTimeout(turnTimeout);
+ turnTimeout = 0;
+ }
+}
+
+/* The single entry point for chat feedback. Every in-flight state has a
+ * visible indicator; every terminal state re-enables the button. */
+export function setUiState(state, errorDetail = "") {
+ uiState = state;
+ stopThinkingClock();
+ clearTurnTimeout(); // the guard only owns the pre-token window
+
+ const inFlight = state === UI_STATE.thinking || state === UI_STATE.streaming;
+ sendBtn.disabled = inFlight;
+ sendBtn.querySelector(".spinner").hidden = !inFlight;
+ sendLabel.textContent = inFlight ? "Thinking…" : "Send";
+ sendStatus.textContent = SEND_STATUS[state] ?? "";
+
+ if (state === UI_STATE.thinking) {
+ addTyping();
+ startThinkingClock();
+ } else {
+ removeTyping();
+ }
+ if (state === UI_STATE.error) showErrorBanner(errorDetail);
}
function autoGrow() {
@@ -247,7 +346,7 @@ function showErrorBanner(detail) {
banner.hidden = false;
banner.classList.add("is-error");
banner.setAttribute("role", "alert");
- bannerText.textContent = `${detail} Try your question again — I'm ready.`;
+ bannerText.textContent = detail ? `${detail} ${ERROR_HINT}` : ERROR_HINT;
}
function clearErrorBanner() {
@@ -268,13 +367,23 @@ async function handleSend(e) {
input.value = "";
autoGrow();
clearErrorBanner();
- setBusy(true);
- addTyping();
let wrap = null;
let acc = "";
let res = null;
+ let aborted = false; // the 120s guard already took the turn to error
+
try {
+ // thinking = pre-token: dots + busy button. The guard is armed so a
+ // hung stream can never leave the button zombified; it clears on the
+ // first delta (entering "streaming") and on every terminal transition.
+ setUiState(UI_STATE.thinking);
+ armTurnTimeout(() => {
+ aborted = true;
+ try { res?.body?.cancel(); } catch { /* already closed */ }
+ setUiState(UI_STATE.error, "That's taking a long time — the answer may be stuck.");
+ });
+
res = await fetch("/api/chat", {
method: "POST",
headers: { "Content-Type": "application/json" },
@@ -289,17 +398,19 @@ async function handleSend(e) {
throw new Error(detail);
}
await readSSE(res, (ev) => {
+ if (aborted) return;
if (ev.type === "delta") {
acc += ev.text || "";
if (!wrap) {
- removeTyping();
+ // First token: dots out, live bubble in; the button stays busy.
+ setUiState(UI_STATE.streaming);
wrap = addMessage("brain", "");
}
wrap.querySelector(".bubble").innerHTML = renderMarkdown(acc);
- wrap.scrollIntoView({ behavior: "smooth", block: "end" });
+ wrap.scrollIntoView({ behavior: SCROLL, block: "end" });
} else if (ev.type === "done") {
if (!wrap) {
- removeTyping();
+ setUiState(UI_STATE.streaming);
wrap = addMessage("brain", "…");
}
if (ev.deflected) {
@@ -311,16 +422,24 @@ async function handleSend(e) {
throw new Error(ev.detail || "Something went wrong on my side.");
}
});
- if (!wrap) {
- removeTyping();
+ if (!aborted && !wrap) {
addMessage("brain", "Hmm — that came back empty. Ask me again?");
}
} catch (err) {
- removeTyping();
- showErrorBanner(err.message || "Something went wrong on my side.");
+ if (!aborted) {
+ const detail =
+ err instanceof Error && err.message
+ ? err.message
+ : "Something went wrong on my side.";
+ setUiState(UI_STATE.error, detail);
+ }
} finally {
+ // done | error → idle: always settle, always focus back. State is
+ // turn-local, so a page reload mid-stream leaves a usable composer.
+ clearTurnTimeout();
+ stopThinkingClock();
try { res?.body?.cancel(); } catch { /* stream already closed */ }
- setBusy(false);
+ if (uiState !== UI_STATE.idle) setUiState(UI_STATE.idle);
input.focus();
}
}
diff --git a/tests/e2e/mock_llm.py b/tests/e2e/mock_llm.py
index d5d585e..338421b 100644
--- a/tests/e2e/mock_llm.py
+++ b/tests/e2e/mock_llm.py
@@ -18,7 +18,10 @@ from __future__ import annotations
import hashlib
import math
+import os
import re
+import signal
+import threading
import time
import uuid
from typing import Any
@@ -79,6 +82,20 @@ def compose_answer(body: dict[str, Any]) -> str:
)
+@app.post("/__shutdown__")
+def shutdown() -> dict[str, Any]:
+ """Test hook (loading-feedback story): terminate this mock process to
+ simulate an LLM outage. The E2E fixture restores a fresh instance on
+ the same port afterwards, so the rest of the session keeps working."""
+
+ def _die() -> None:
+ time.sleep(0.1) # let the HTTP response flush before we exit
+ os.kill(os.getpid(), signal.SIGTERM)
+
+ threading.Thread(target=_die, daemon=True).start()
+ return {"status": "shutting down"}
+
+
@app.get("/v1/models")
def models() -> dict[str, Any]:
return {
diff --git a/tests/e2e/test_loading_feedback.py b/tests/e2e/test_loading_feedback.py
new file mode 100644
index 0000000..b1422ab
--- /dev/null
+++ b/tests/e2e/test_loading_feedback.py
@@ -0,0 +1,311 @@
+"""Phase 06 E2E (Playwright): loading feedback & progress.
+
+Story: ``.agent/user_stories/loading-feedback.md``
+Run in isolation (DB must be up: ``podman compose up -d db``):
+
+ uv run pytest tests/e2e/test_loading_feedback.py -v --no-cov
+
+Determinism comes from two mock-LLM behaviors (tests/e2e/mock_llm.py):
+
+* ``pretend to think slowly`` in the user message → a 3s warm-up before
+ the first token, wide enough to assert the pre-token UI (typing dots +
+ busy "Thinking…" button) at a known timestamp;
+* the ``POST /__shutdown__`` hook → the ``llm_down`` fixture stops the
+ shared mock to simulate an LLM outage, then restores a fresh instance
+ on the same port so later tests keep working.
+
+The state machine under test lives in frontend/assets/app.js:
+``idle → thinking → streaming → done | error → idle`` (PLAN §7.4).
+"""
+from __future__ import annotations
+
+import asyncio
+import os
+import re
+import subprocess
+import sys
+import time
+from collections.abc import Iterator
+from pathlib import Path
+from threading import Thread
+from typing import Any
+
+import httpx
+import pytest
+from playwright.sync_api import Page, expect
+from sqlalchemy import text
+
+from app.config import Settings
+from app.db import SessionLocal
+from app.rag.importer import ImportSummary, import_sources
+from app.rag.llm import LLMClient
+
+REPO = Path(__file__).resolve().parents[2]
+FIXTURES = REPO / "tests" / "fixtures" / "docs"
+MOCK_PORT = int(os.environ.get("E2E_MOCK_PORT", "8901"))
+
+# The mock warms up for 3s when the question contains this marker, so the
+# pre-token window is observable. (Retrieval may answer or deflect — the
+# feedback states are identical either way.)
+SLOW_QUESTION = "pretend to think slowly then tell me about kubernetes"
+ON_TOPIC = "How is my Kubernetes cluster set up?"
+
+TYPING = "#typing-indicator"
+# The typing indicator is itself a .msg.brain — exclude its bubble.
+ANSWER = ".msg.brain .bubble:not(.typing)"
+
+
+# --------------------------------------------------------------------------
+# KB seeding (same pattern as the phase 02/03 story suites)
+# --------------------------------------------------------------------------
+
+
+async def _import_fixtures(mock_port: int) -> ImportSummary:
+ kwargs: dict[str, Any] = {"_env_file": None, "llm_base_url": f"http://127.0.0.1:{mock_port}/v1"}
+ settings = Settings(**kwargs) # pyright: ignore[reportCallIssue]
+ return await import_sources([FIXTURES], LLMClient(settings))
+
+
+def _run_in_thread(coro: Any) -> Any:
+ """Run a coroutine on a worker thread.
+
+ Playwright's sync API keeps an asyncio loop running on the test thread,
+ so ``asyncio.run`` cannot be called directly from a test body.
+ """
+ box: dict[str, Any] = {}
+
+ def runner() -> None:
+ try:
+ box["value"] = asyncio.run(coro)
+ except BaseException as e: # noqa: BLE001 — re-raised on the test thread
+ box["error"] = e
+
+ t = Thread(target=runner)
+ t.start()
+ t.join()
+ if "error" in box:
+ raise box["error"]
+ return box["value"]
+
+
+def _reset_db(mock_port: int, seed: bool) -> ImportSummary | None:
+ """Truncate the KB (and query log), then optionally re-import fixtures."""
+ with SessionLocal() as db:
+ db.execute(text("TRUNCATE chunks, documents, query_log"))
+ db.commit()
+ if not seed:
+ return None
+ return _run_in_thread(_import_fixtures(mock_port))
+
+
+# --------------------------------------------------------------------------
+# LLM-down fixture: stop the shared mock, restore it for later tests
+# --------------------------------------------------------------------------
+
+_replacement_mocks: list[subprocess.Popen] = []
+
+
+@pytest.fixture(scope="session")
+def _cleanup_replacement_mocks() -> Iterator[None]:
+ """Terminate mock instances spawned to replace a stopped one."""
+ yield
+ for proc in _replacement_mocks:
+ proc.terminate()
+ try:
+ proc.wait(timeout=10)
+ except subprocess.TimeoutExpired:
+ proc.kill()
+
+
+def _wait_http(url: str, timeout: float = 40.0) -> None:
+ deadline = time.monotonic() + timeout
+ while time.monotonic() < deadline:
+ try:
+ httpx.get(url, timeout=2.0)
+ return
+ except Exception: # noqa: BLE001 — retry until deadline
+ time.sleep(0.5)
+ raise RuntimeError(f"server at {url} did not come up")
+
+
+def _spawn_mock() -> subprocess.Popen:
+ env = dict(os.environ)
+ env.pop("DEBUGPY", None)
+ return subprocess.Popen(
+ [sys.executable, "-m", "uvicorn", "tests.e2e.mock_llm:app",
+ "--host", "127.0.0.1", "--port", str(MOCK_PORT), "--log-level", "warning"],
+ cwd=REPO,
+ env=env,
+ )
+
+
+@pytest.fixture()
+def llm_down(mock_llm: int, _cleanup_replacement_mocks: None) -> Iterator[None]:
+ """Simulate the LLM going down for one test (stop the shared mock via
+ its ``__shutdown__`` hook), then restore a fresh instance on the same
+ port so the rest of the session keeps working."""
+ r = httpx.post(f"http://127.0.0.1:{MOCK_PORT}/__shutdown__", timeout=10)
+ assert r.status_code == 200
+
+ stopped = False
+ deadline = time.monotonic() + 15
+ while time.monotonic() < deadline:
+ try:
+ httpx.get(f"http://127.0.0.1:{MOCK_PORT}/v1/models", timeout=1.0)
+ except Exception: # noqa: BLE001 — connection refused == stopped
+ stopped = True
+ break
+ time.sleep(0.2)
+ assert stopped, "mock LLM did not stop in time"
+
+ yield
+
+ proc = _spawn_mock()
+ _replacement_mocks.append(proc)
+ _wait_http(f"http://127.0.0.1:{MOCK_PORT}/v1/models")
+
+
+# --------------------------------------------------------------------------
+# Tests (story → test mapping, .agent/user_stories/loading-feedback.md)
+# --------------------------------------------------------------------------
+
+
+def test_typing_indicator_during_slow_think(
+ page: Page, app_url: str, mock_llm: int, db_ready: None
+) -> None:
+ """AC1/AC5: the 3s mock warm-up must show the typing indicator for
+ >=2s before any text appears, then it is gone once the answer lands."""
+ summary = _reset_db(mock_llm, seed=True)
+ assert summary is not None and summary.added == 3
+ page.set_default_timeout(30_000)
+ page.goto(app_url)
+
+ page.fill("#message-input", SLOW_QUESTION)
+ page.click("#send-btn")
+
+ # Visible within 500ms of submit — the indicator is added synchronously
+ # by setUiState("thinking").
+ typing = page.locator(TYPING)
+ expect(typing).to_be_visible(timeout=500)
+ bubble = typing.locator(".bubble")
+ expect(bubble).to_have_attribute("role", "status")
+ expect(bubble).to_have_attribute("aria-label", "Brain of Reese is thinking")
+
+ # ~2s in: still pre-token (the mock is in its 3s warm-up).
+ page.wait_for_timeout(2000)
+ assert typing.is_visible(), "typing indicator must persist through the pre-token window"
+
+ # First token lands (~3s): dots gone, answer text present.
+ answer = page.locator(ANSWER)
+ answer.wait_for(state="visible", timeout=30_000)
+ expect(typing).to_be_hidden()
+ assert answer.inner_text().strip(), "answer bubble must contain text"
+
+
+def test_button_state_machine(
+ page: Page, app_url: str, mock_llm: int, db_ready: None
+) -> None:
+ """AC1/AC3: disabled + spinner + 'Thinking…' while in flight; enabled
+ + 'Send' + focused input after done."""
+ _reset_db(mock_llm, seed=True)
+ page.set_default_timeout(30_000)
+ page.goto(app_url)
+
+ page.fill("#message-input", SLOW_QUESTION)
+ page.click("#send-btn")
+
+ btn = page.locator("#send-btn")
+ label = page.locator("#send-label")
+ expect(btn).to_be_disabled(timeout=500)
+ expect(label).to_have_text("Thinking…")
+ expect(btn.locator(".spinner")).to_be_visible()
+ expect(page.locator("#send-status")).to_contain_text("thinking")
+
+ # Done: button recovers and the input is focused back.
+ page.locator(ANSWER).wait_for(state="visible", timeout=30_000)
+ expect(label).to_have_text("Send", timeout=30_000)
+ expect(btn).to_be_enabled()
+ expect(btn.locator(".spinner")).to_be_hidden()
+ expect(page.locator("#message-input")).to_be_focused()
+
+
+def test_streaming_appends_live(
+ page: Page, app_url: str, mock_llm: int, db_ready: None
+) -> None:
+ """AC2: text appends live — a sample taken mid-stream must be strictly
+ shorter than a later one (no 'whole answer appears at once')."""
+ _reset_db(mock_llm, seed=True)
+ page.set_default_timeout(30_000)
+ page.goto(app_url)
+
+ page.fill("#message-input", ON_TOPIC)
+ page.click("#send-btn")
+
+ answer = page.locator(ANSWER)
+ answer.wait_for(state="visible", timeout=30_000)
+ first = answer.inner_text()
+ assert first.strip(), "bubble should carry text as soon as it appears"
+
+ second = first
+ deadline = time.monotonic() + 15
+ while time.monotonic() < deadline:
+ second = answer.inner_text()
+ if len(second) > len(first):
+ break
+ time.sleep(0.05)
+ assert len(second) > len(first), (
+ "answer text never grew after the first sample — no live streaming visible"
+ )
+
+ # The turn settles: button back to 'Send', final text no shorter.
+ expect(page.locator("#send-label")).to_have_text("Send", timeout=30_000)
+ final = answer.inner_text()
+ assert len(final) >= len(second)
+
+
+def test_reduced_motion_keeps_feedback(
+ page: Page, app_url: str, mock_llm: int, db_ready: None
+) -> None:
+ """AC7: under prefers-reduced-motion the dots/spinner stay visible
+ (static/slower) — feedback is never removed, only calmed."""
+ _reset_db(mock_llm, seed=True)
+ page.emulate_media(reduced_motion="reduce")
+ page.set_default_timeout(30_000)
+ page.goto(app_url)
+
+ page.fill("#message-input", SLOW_QUESTION)
+ page.click("#send-btn")
+
+ expect(page.locator(TYPING)).to_be_visible(timeout=500)
+ expect(page.locator(TYPING).locator(".bubble span")).to_have_count(3)
+ expect(page.locator("#send-btn .spinner")).to_be_visible()
+
+ # And the turn still completes and recovers.
+ expect(page.locator("#send-label")).to_have_text("Send", timeout=30_000)
+ expect(page.locator("#send-btn")).to_be_enabled()
+ expect(page.locator(TYPING)).to_be_hidden()
+
+
+def test_error_banner_on_llm_down(
+ page: Page, app_url: str, mock_llm: int, db_ready: None, llm_down: None
+) -> None:
+ """AC4a: with the LLM down the turn ends in the red role=alert banner
+ with the actionable retry hint, and the button recovers (never a
+ zombie)."""
+ page.set_default_timeout(30_000)
+ page.goto(app_url)
+
+ page.fill("#message-input", ON_TOPIC)
+ page.click("#send-btn")
+
+ banner = page.locator("#kb-banner")
+ expect(banner).to_be_visible(timeout=30_000)
+ expect(banner).to_have_attribute("role", "alert")
+ expect(banner).to_have_class(re.compile(r"kb-banner.*is-error"))
+ expect(banner).to_contain_text("Try again")
+ expect(banner).to_contain_text("check the LLM is reachable")
+
+ # Recovery: button enabled, label 'Send', no lingering indicator.
+ expect(page.locator("#send-btn")).to_be_enabled(timeout=30_000)
+ expect(page.locator("#send-label")).to_have_text("Send")
+ expect(page.locator(TYPING)).to_be_hidden()
diff --git a/tests/integration/test_chat_api.py b/tests/integration/test_chat_api.py
index 7f1fd41..ac83a86 100644
--- a/tests/integration/test_chat_api.py
+++ b/tests/integration/test_chat_api.py
@@ -264,6 +264,24 @@ def test_chat_mid_stream_failure_yields_error_after_partial_deltas(client, db) -
assert db.scalars(select(QueryLog)).all() == []
+def test_error_event_matches_contract_shape(client, db, seeded_kb) -> None:
+ """The SSE error event (PLAN §4) is exactly ``{type, detail}`` — the
+ client's loading-feedback state machine (phase 06) keys off this shape
+ to flip to the error state and re-enable the send button."""
+ broken = FakeRagLLM(embed_error=EmbeddingError("embeddings endpoint down"))
+ fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: broken
+ try:
+ _, _, frames = _stream_chat(client, QUESTION)
+ finally:
+ fastapi_app.dependency_overrides.clear()
+
+ assert len(frames) == 1
+ event = frames[0]
+ assert set(event.keys()) == {"type", "detail"}
+ assert event["type"] == "error"
+ assert isinstance(event["detail"], str) and event["detail"]
+
+
def test_chat_db_down_returns_503_json(client, monkeypatch) -> None:
monkeypatch.setattr(chat_api, "db_available", lambda: False)
r = client.post("/api/chat", json={"message": "hello"})
diff --git a/tests/unit/test_frontend_feedback.py b/tests/unit/test_frontend_feedback.py
new file mode 100644
index 0000000..637173e
--- /dev/null
+++ b/tests/unit/test_frontend_feedback.py
@@ -0,0 +1,93 @@
+"""Unit: the loading-feedback contract in the static frontend (phase 06).
+
+The JS behavior itself is E2E-covered (tests/e2e/test_loading_feedback.py);
+here we pin the exported constants and state-machine markers that the
+story depends on, so a silent regression in app.js/styles.css is caught
+without a browser.
+"""
+from __future__ import annotations
+
+import re
+from pathlib import Path
+
+FRONTEND = Path(__file__).resolve().parents[2] / "frontend"
+APP_JS = FRONTEND / "assets" / "app.js"
+STYLES_CSS = FRONTEND / "assets" / "styles.css"
+
+
+def _js() -> str:
+ return APP_JS.read_text(encoding="utf-8")
+
+
+def _css() -> str:
+ return STYLES_CSS.read_text(encoding="utf-8")
+
+
+def test_turn_timeout_constant_exported_at_120s() -> None:
+ """The 120s client-side guard (PLAN §7.4) must be an *exported*
+ constant — testable, and the single value the E2E timeout story keys
+ off."""
+ js = _js()
+ match = re.search(r"export\s+const\s+TURN_TIMEOUT_MS\s*=\s*120_?000\s*;", js)
+ assert match, "app.js must export `const TURN_TIMEOUT_MS = 120000`"
+
+
+def test_state_machine_has_all_four_states() -> None:
+ """idle → thinking → streaming → done | error → idle (PLAN §7.4).
+
+ `done` is not a UI state: it settles into `idle` in the turn handler's
+ finally block, so the state machine itself has exactly four states.
+ """
+ js = _js()
+ for state in ("idle", "thinking", "streaming", "error"):
+ assert re.search(rf'{state}:\s*"{state}"', js), f"state {state!r} missing"
+ assert "function setUiState" in js, "setUiState must exist as the single entry point"
+ assert "export function setUiState" in js, "setUiState must be exported (testable)"
+
+
+def test_typing_indicator_contract_strings() -> None:
+ """The indicator's accessible label and the 10s elapsed-seconds hint
+ are the strings screen readers (and the E2E) rely on."""
+ js = _js()
+ assert 'TYPING_LABEL = "Brain of Reese is thinking"' in js
+ assert "still thinking" in js, "elapsed-seconds hint must update the aria label"
+ assert "secs < 10" in js, "the hint must only appear after 10s of silence"
+ assert "role=\"status\"" in js, "typing bubble must be role=status"
+
+
+def test_error_banner_has_actionable_hint() -> None:
+ """Every error path (SSE error event, non-2xx, 120s timeout) must
+ surface the same actionable retry hint (story AC4)."""
+ js = _js()
+ assert "check the LLM is reachable" in js
+ assert '"role", "alert"' in js, "error banner must be role=alert"
+ # the banner must be the red error variant
+ assert "is-error" in js
+
+
+def test_reduced_motion_calm_not_removed() -> None:
+ """prefers-reduced-motion: feedback is calmed, never removed (story
+ AC7). Dots go static; the spinner only slows down."""
+ css = _css()
+ blocks = re.findall(
+ r"@media \(prefers-reduced-motion: reduce\) \{([\s\S]*?)\n\}", css
+ )
+ assert blocks, "styles.css must contain prefers-reduced-motion rules"
+ assert any(".typing span" in b and "animation: none" in b for b in blocks), (
+ "typing dots must have a static fallback under reduced motion"
+ )
+ assert any(".spinner" in b and "animation-duration" in b for b in blocks), (
+ "spinner must slow down (not vanish) under reduced motion"
+ )
+
+
+def test_busy_button_style_tokens() -> None:
+ """Story spec: busy send button is #a5b4fc with the 16px white-arc
+ spinner; the label swaps Send ↔ Thinking…."""
+ css = _css()
+ js = _js()
+ assert ".send-btn:disabled" in css
+ assert "#a5b4fc" in css
+ assert re.search(r"\.spinner \{[^}]*width: 16px", css)
+ assert "Thinking…" in js
+ assert 'sendLabel.textContent' in js
diff --git a/tests/unit/test_sse_events.py b/tests/unit/test_sse_events.py
index 23449f8..f63cf1f 100644
--- a/tests/unit/test_sse_events.py
+++ b/tests/unit/test_sse_events.py
@@ -4,6 +4,7 @@ from __future__ import annotations
import json
from app.api.chat import sse_event
+from app.schemas import ChatErrorEvent
def _payload(frame: str) -> dict:
@@ -46,3 +47,17 @@ def test_multi_line_text_stays_one_frame() -> None:
frame = sse_event({"type": "delta", "text": "line1\nline2\n\n"})
assert frame.count("\n\n") == 1 # only the frame terminator
assert _payload(frame)["text"] == "line1\nline2\n\n"
+
+
+def test_error_event_model_serializes_exact_frame() -> None:
+ """The ``ChatErrorEvent`` model is the wire shape of every server-side
+ failure the UI's state machine (phase 06) must recover from."""
+ frame = sse_event(ChatErrorEvent(detail="boom").model_dump())
+ assert frame == 'data: {"type": "error", "detail": "boom"}\n\n'
+ assert _payload(frame) == {"type": "error", "detail": "boom"}
+
+
+def test_error_event_shape_is_type_and_detail_only() -> None:
+ dumped = ChatErrorEvent(detail="The chat model dropped the connection").model_dump()
+ assert set(dumped.keys()) == {"type", "detail"}
+ assert dumped["type"] == "error" # default — call sites never spell it out