From da9fefdbd243877f12f2f0e253355996513aa926 Mon Sep 17 00:00:00 2001 From: agrechenkov Date: Mon, 27 Jul 2026 14:07:35 -0400 Subject: [PATCH 1/2] Stream Chat tab replies via SSE (prompt-mode plan Phase 3) The dashboard Chat tab now renders the assistant reply token-by-token over a text/event-stream response instead of a blank "Thinking..." wait. First token arrives at ~1.6s warm (measured live on FastFlowLM qwen3.5:4b) instead of after the whole completion. - ffp_chat: stream_send() generator (delta -> done/error terminal), _default_llm_stream + _parse_sse_delta SSE parser; send() refactored onto shared _build_send_context/_persist_turn. Persistence re-reads the thread store under the write-lock so a concurrent delete/append during a multi-second stream is never clobbered by a stale snapshot, and a client disconnect (GeneratorExit) still persists the partial reply. - ffp_daemon: _STREAM_ACTIONS + _sse_frame + Handler._stream_action; do_POST streams event-stream frames while keeping the X-FFP-API CSRF + Host gates. - app.js: streamChat() fetch-body reader + parseSseFrame(); sendChat() streams with a transparent fallback to the one-shot chat_send action on any older daemon or open failure. - AHK paste path + grammar/prompt modes untouched (whole-output). Works on FastFlowLM and Ollama (shared OpenAI-compatible SSE). - 25 new tests: SSE parse, stream_send accumulate/persist/notes/partial-on- error, concurrency lost-update guard, client-disconnect persist, daemon end-to-end SSE + CSRF + disconnect. Fixes found by an adversarial review pass. Phase 4 (faster-decode route) deferred: its plan trigger (p50 > 15s) is already resolved by 2.3.0 (3.38s warm) and streaming delivers the perceived speed it targeted. SPEC V37/T27; CHANGELOG Unreleased. Co-Authored-By: Claude Opus 4.8 --- CHANGELOG.md | 4 + SPEC.md | 2 + docs/prompt-mode-speed-quality-plan.md | 29 +++- scripts/ffp_chat.py | 220 +++++++++++++++++++++---- scripts/ffp_daemon.py | 84 +++++++++- scripts/ui/web/app.js | 100 +++++++++-- tests/test_ffp_chat_stream.py | 196 ++++++++++++++++++++++ tests/test_ffp_daemon.py | 134 +++++++++++++++ 8 files changed, 716 insertions(+), 53 deletions(-) create mode 100644 tests/test_ffp_chat_stream.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 43e81a0..0ea36e0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,10 @@ ## Unreleased +### Added + +- **The Chat tab streams replies as they generate.** Instead of a blank "Thinking…" wait until the whole answer lands, the assistant's reply now fills in token-by-token over a Server-Sent-Events stream — first text typically appears at ~1.6 s (warm TTFT) rather than after the full completion. Prompt-mode/grammar hotkeys and the AHK paste path are unchanged (they still return whole output). A daemon without the streaming endpoint, or any failure to open the stream, transparently falls back to the previous one-shot request, so nothing regresses. Works on both FastFlowLM and Ollama (shared OpenAI-compatible SSE). A mid-stream provider drop still saves the partial reply and shows the error. + ## 2.3.0 **Prompt mode, faster and better grounded.** The default `prompt:` path now emits a compact copy-paste-ready agent prompt in a few seconds without filling gaps with conventional-but-unstated requirements. diff --git a/SPEC.md b/SPEC.md index 2f60670..8f3aedf 100644 --- a/SPEC.md +++ b/SPEC.md @@ -88,6 +88,7 @@ Caveman-encoded (compression, not amputation). Paths / ids / action names / numb - V34: FastFlowLM force restart → old port observed closed before new spawn; ⊥ return `already_running` from dying instance - V35: default v2 → exactly 1 short LLM draft call; ⊥ anti-echo/rescue calls; V33 finalizer supplies surfaced 4-section contract - V36: pre-2.3 prompt_builder identity cfg (⊥ `prompt_version`, legacy default fields) → migrate v2+concise; any custom field → preserve +- V37: `chat_send_stream` streams reply deltas as `text/event-stream` (SSE `data:`/`event:` frames), persists the full-or-partial turn exactly once under the daemon write-lock; CSRF `X-FFP-API` + Host gates still apply; dashboard falls back to `chat_send` iff the stream ⊥ opens (never after tokens arrive); `send()`/AHK/grammar/prompt paths unchanged (⊥ streamed) ## §T tasks @@ -119,6 +120,7 @@ T23|x|prompt-v2 fixed A/B speed+quality harness + tests|V29 T24|x|prompt-v2 default + v1 rollback selector + 240/320/420 caps|V24,V27,V28,V30,V32 T25|x|FastFlowLM startup+idle keep-warm + cold/warm measurement support|V31 T26|x|2.3.0 release evidence + version/docs rebaseline after A/B gate passes|V18,V29,V33,V34,V35,V36 +T27|x|streaming Chat tab (SSE): `ffp_chat.stream_send` + daemon `_STREAM_ACTIONS`/`_sse_frame`/`_stream_action` + `app.js` fetch-reader w/ `chat_send` fallback; live FLM first-token 1.58s|V37 ``` ## §B bugs diff --git a/docs/prompt-mode-speed-quality-plan.md b/docs/prompt-mode-speed-quality-plan.md index 3d39ea8..730a1de 100644 --- a/docs/prompt-mode-speed-quality-plan.md +++ b/docs/prompt-mode-speed-quality-plan.md @@ -1,8 +1,8 @@ # Prompt Mode Speed + Quality Refresh Plan ("prompt v2") -Date: 2026-07-09 -Status: PLAN ONLY — do not implement. Includes measured baseline + the exact -measurement harness to build. +Date: 2026-07-09 (Phases 0–2 shipped in 2.3.0; Phase 3 done post-2.3.0; Phase 4 deferred) +Status: Phases 0–2 SHIPPED (2.3.0). Phase 3 (streaming Chat tab) DONE on +`feat/streaming-chat`. Phase 4 (faster route) trigger not met → deferred. Target release: **2.3.0** (re-baselines the default `prompt:` output — see Risk) ## Problem @@ -143,11 +143,24 @@ like the 2.2.0 prompt-builder eval. and likely hits the goal alone. - **Phase 2 — Warm-model guarantee.** Ensure FLM stays loaded so nobody eats the cold-load tax; verify with a cold-vs-warm measurement. -- **Phase 3 (optional) — Streaming.** SSE from daemon → dashboard Chat/preview - for perceived latency; AHK paste path stays whole-output. -- **Phase 4 (optional) — Faster route.** Only if p50 still > 15 s after 1–2: - evaluate an iGPU/LM-Studio decode route behind the provider split, gated on the - same quality eval. +- **Phase 3 (optional) — Streaming. ✅ DONE (post-2.3.0, `feat/streaming-chat`).** + The surface that actually benefits is the **Chat tab** (replies up to 1024 + tokens, formerly a blank wait) — `prompt:` has no live dashboard surface and the + AHK paste path pastes at the end, so streaming there is moot. Delivered as a + `text/event-stream` response over a `fetch()` POST (keeps the `X-FFP-API` CSRF + gate that `EventSource` cannot send): `ffp_chat.stream_send` generator → + daemon `_STREAM_ACTIONS`/`_sse_frame`/`_stream_action` → `app.js` fetch-reader, + with a transparent fallback to the one-shot `chat_send` on any older daemon. + Measured live on FastFlowLM `qwen3.5:4b`: first token **1.58 s** vs **2.10 s** + total — text flows in instead of a blank wait. AHK/grammar/prompt paths + untouched. See SPEC V37 / T27. +- **Phase 4 (optional) — Faster route. ⏸ TRIGGER NOT MET — deferred.** The plan + gates this on "p50 still > 15 s after phases 1–2," but 2.3.0 put prompt p50 at + **3.38 s warm** and Phase 3 streaming already delivers the *perceived* speed + this phase was chasing. A second NPU-decode provider (LM Studio/Lemonade) also + can't co-serve the NPU with FastFlowLM (both wedge — see the benchmark + evidence), so it would add real complexity for no measured need. Revisit only + if a concrete slow surface appears. ## Integration with the existing prompt-builder (2.2.0) diff --git a/scripts/ffp_chat.py b/scripts/ffp_chat.py index 32bd1a4..c0a6843 100644 --- a/scripts/ffp_chat.py +++ b/scripts/ffp_chat.py @@ -10,12 +10,14 @@ from __future__ import annotations +import contextlib import json import logging import time import urllib.error import urllib.request import uuid +from collections.abc import Iterator import paths as _paths @@ -240,6 +242,69 @@ def _default_llm_call(messages: list[dict]) -> str: return str((choices[0].get("message") or {}).get("content") or "").strip() +def _parse_sse_delta(raw_line: str | bytes) -> Iterator[str]: + """Yield the assistant text delta(s) carried by one OpenAI-style SSE line. + + The streaming ``/v1/chat/completions`` response is a sequence of + ``data: {json}`` lines (plus blank separators and a final ``data: [DONE]``), + identical on FastFlowLM and Ollama's OpenAI-compatible endpoint. Non-data + lines, the terminator, and malformed JSON yield nothing, so a partial or + noisy stream degrades to fewer tokens rather than an exception. + """ + line = raw_line.decode("utf-8", errors="replace") if isinstance(raw_line, bytes) else str(raw_line) + line = line.strip() + if not line or not line.startswith("data:"): + return + data = line[len("data:"):].strip() + if not data or data == "[DONE]": + return + try: + obj = json.loads(data) + except (ValueError, TypeError): + return + for choice in obj.get("choices") or []: + delta = (choice.get("delta") or {}).get("content") + if delta: + yield str(delta) + + +def _default_llm_stream(messages: list[dict]) -> Iterator[str]: + """Yield reply text deltas from the active provider's streaming chat endpoint. + + Mirrors :func:`_default_llm_call` (same provider-resolved endpoint/model/auth + from ``grammar_fix``) but sets ``stream: True`` and iterates the SSE body so + the daemon can forward tokens to the dashboard as they arrive. Raises + RuntimeError on transport failure; a mid-stream drop simply ends iteration. + """ + import grammar_fix + + base_url = str(getattr(grammar_fix, "FLM_BASE_URL", "http://127.0.0.1:52625") or "").rstrip("/") + model = str(getattr(grammar_fix, "FLM_MODEL", "qwen3.5:4b") or "qwen3.5:4b") + bearer = str(getattr(grammar_fix, "LLM_AUTH_BEARER", "flm") or "") + timeout = int(getattr(grammar_fix, "FLM_TIMEOUT_SECONDS", 240) or 240) + + body = json.dumps( + { + "model": model, + "messages": messages, + "temperature": TEMPERATURE, + "max_tokens": MAX_TOKENS, + "stream": True, + } + ).encode("utf-8") + headers = {"Content-Type": "application/json", "Accept": "text/event-stream"} + if bearer: + headers["Authorization"] = f"Bearer {bearer}" + req = urllib.request.Request(base_url + "/v1/chat/completions", data=body, headers=headers, method="POST") + try: + resp = urllib.request.urlopen(req, timeout=max(2, timeout)) + except urllib.error.URLError as e: + raise RuntimeError(f"LLM unreachable at {base_url}: {getattr(e, 'reason', e)}") from e + with resp: + for raw in resp: # http.client yields lines as they flush off the socket + yield from _parse_sse_delta(raw) + + def _derive_title(text: str) -> str: text = (text or "").strip() if not text: @@ -250,28 +315,24 @@ def _derive_title(text: str) -> str: return (first[:TITLE_MAX_CHARS] + "…") if len(first) > TITLE_MAX_CHARS else first -def send(thread_id: str = "", message: str = "", use_notes: bool = False, llm_call=None) -> dict: - """Append the user message, call the LLM (optionally grounded in the notes - vault), persist, and return ``{thread_id, title, reply, notes_used}``. +def _build_send_context(thread_id: str, message: str, use_notes: bool) -> tuple[str, list[dict], list[str]]: + """Resolve the target thread id and assemble the model request. - ``llm_call(messages)->str`` is injectable for tests; defaults to the live - provider endpoint. + Returns ``(thread_id, messages, notes_used)`` — the resolved/new thread id, + the system+grounding+window+turn message list, and the grounding titles. + Only *reads* the thread store (for the sliding-window history); the write + happens later in :func:`_persist_turn`, which re-reads under the lock so the + read-here/write-later split can never clobber a concurrent write. Shared by + :func:`send` and :func:`stream_send` so both build identical requests. """ - message = str(message or "").strip() - if not message: - raise ValueError("empty chat message") - llm_call = llm_call or _default_llm_call - threads = load_threads() idx = _thread_index(threads, str(thread_id or "")) if thread_id else -1 if idx < 0: thread_id = str(thread_id or "") or uuid.uuid4().hex - thread = {"thread_id": thread_id, "title": "New chat", "updated_at": _now_iso(), "history": []} - threads.insert(0, thread) - idx = 0 - thread = threads[idx] - thread_id = thread.get("thread_id") - history: list[dict] = list(thread.get("history") or []) + history: list[dict] = [] + else: + thread_id = threads[idx].get("thread_id") + history = list(threads[idx].get("history") or []) # Build the request: system + optional notes grounding + sliding window + new turn. messages: list[dict] = [{"role": "system", "content": SYSTEM_PROMPT}] @@ -281,23 +342,126 @@ def send(thread_id: str = "", message: str = "", use_notes: bool = False, llm_ca if ctx: messages.append({"role": "system", "content": ctx}) notes_used = titles - window = history[-(CONTEXT_WINDOW_TURNS * 2):] - for m in window: + for m in history[-(CONTEXT_WINDOW_TURNS * 2):]: role = str(m.get("role") or "") content = str(m.get("content") or "") if role in ("user", "assistant") and content: messages.append({"role": role, "content": content}) messages.append({"role": "user", "content": message}) + return thread_id, messages, notes_used + +def _persist_turn(thread_id: str, message: str, reply: str, save_lock=None) -> str: + """Append the user+assistant turn to ``thread_id`` and save; return its title. + + The whole read-modify-write is wrapped in ``save_lock`` so it is atomic + against concurrent writers: it re-loads the current thread list *inside* the + lock (rather than trusting a snapshot captured before a multi-second stream), + resolves the thread by id, appends the turn, and rewrites. Without the + re-read, a stream that started before a concurrent delete/append would + overwrite the file with its stale snapshot and silently lose that write. The + one-shot ``chat_send`` action passes no lock because ``do_POST`` already holds + ``_write_lock`` across its whole handler; streaming passes the daemon's + ``_write_lock`` here because it runs outside that wrapper. + """ + with (save_lock or contextlib.nullcontext()): + threads = load_threads() + idx = _thread_index(threads, str(thread_id or "")) + if idx < 0: + thread = {"thread_id": thread_id, "title": "New chat", "updated_at": _now_iso(), "history": []} + threads.insert(0, thread) + idx = 0 + thread = threads[idx] + history: list[dict] = list(thread.get("history") or []) + history.append({"role": "user", "content": message}) + history.append({"role": "assistant", "content": reply}) + if not thread.get("title") or thread.get("title") == "New chat": + thread["title"] = _derive_title(message) + thread["history"] = history + thread["updated_at"] = _now_iso() + threads.pop(idx) + threads.insert(0, thread) + save_threads(threads[:MAX_THREADS]) + return thread["title"] + + +def send(thread_id: str = "", message: str = "", use_notes: bool = False, llm_call=None) -> dict: + """Append the user message, call the LLM (optionally grounded in the notes + vault), persist, and return ``{thread_id, title, reply, notes_used}``. + + ``llm_call(messages)->str`` is injectable for tests; defaults to the live + provider endpoint. + """ + message = str(message or "").strip() + if not message: + raise ValueError("empty chat message") + llm_call = llm_call or _default_llm_call + + thread_id, messages, notes_used = _build_send_context(thread_id, message, use_notes) reply = str(llm_call(messages) or "").strip() + title = _persist_turn(thread_id, message, reply) + return {"thread_id": thread_id, "title": title, "reply": reply, "notes_used": notes_used} + + +def stream_send( + thread_id: str = "", + message: str = "", + use_notes: bool = False, + llm_stream=None, + save_lock=None, +) -> Iterator[dict]: + """Streaming twin of :func:`send`: yield reply deltas, then a terminal event. + + Yields ``{"type": "delta", "delta": text}`` for each token chunk as it + arrives, then exactly one terminal event: + ``{"type": "done", thread_id, title, notes_used}`` on success, or + ``{"type": "error", error, thread_id, title, notes_used}`` if the provider + stream failed mid-flight. Whatever text arrived before a failure — a provider + drop *or* a client disconnect (``GeneratorExit`` thrown in when the daemon + stops iterating) — is still persisted, so the transcript never loses a + partial reply. + + ``llm_stream(messages) -> Iterator[str]`` is injectable for tests; defaults + to the live provider's streaming endpoint. ``save_lock`` serializes the final + write with the daemon's other writers. + """ + message = str(message or "").strip() + if not message: + raise ValueError("empty chat message") + llm_stream = llm_stream or _default_llm_stream - history.append({"role": "user", "content": message}) - history.append({"role": "assistant", "content": reply}) - if not thread.get("title") or thread.get("title") == "New chat": - thread["title"] = _derive_title(message) - thread["history"] = history - thread["updated_at"] = _now_iso() - threads.pop(idx) - threads.insert(0, thread) - save_threads(threads[:MAX_THREADS]) - return {"thread_id": thread_id, "title": thread["title"], "reply": reply, "notes_used": notes_used} + thread_id, messages, notes_used = _build_send_context(thread_id, message, use_notes) + parts: list[str] = [] + err: Exception | None = None + try: + for delta in llm_stream(messages): + delta = str(delta or "") + if not delta: + continue + parts.append(delta) + yield {"type": "delta", "delta": delta} + except GeneratorExit: + # The daemon abandoned the stream (client disconnected). We cannot yield + # anymore, but we still persist the partial reply — matching the contract + # and the non-streaming path — then let the generator close. + partial = "".join(parts).strip() + if partial: + with contextlib.suppress(Exception): + _persist_turn(thread_id, message, partial, save_lock=save_lock) + raise + except Exception as exc: # transport drop / provider error mid-stream + err = exc + log.warning("chat stream failed after %d chunk(s): %s", len(parts), exc) + + reply = "".join(parts).strip() + title = _persist_turn(thread_id, message, reply, save_lock=save_lock) if reply else "New chat" + + terminal = { + "type": "error" if err else "done", + "thread_id": thread_id, + "title": title, + "notes_used": notes_used, + } + if err: + terminal["error"] = str(err) or "chat stream failed" + yield terminal diff --git a/scripts/ffp_daemon.py b/scripts/ffp_daemon.py index cfa7605..b268e34 100644 --- a/scripts/ffp_daemon.py +++ b/scripts/ffp_daemon.py @@ -18,6 +18,7 @@ from __future__ import annotations import argparse +import contextlib import json import logging import logging.handlers @@ -27,7 +28,7 @@ import threading import time import traceback -from collections.abc import Callable +from collections.abc import Callable, Iterator from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any @@ -829,6 +830,37 @@ def _act_meeting_week_summary(args: dict) -> dict: "notify_gate", # writes the notifications log + updates dedupe state } +def _stream_chat_send(args: dict) -> Iterator[dict]: + """Streaming counterpart of ``chat_send``: yields delta/done/error events. + + The final write is serialized with the daemon ``_write_lock`` (streaming is + handled outside the do_POST write-lock wrapper, so it passes the lock down).""" + import ffp_chat + return ffp_chat.stream_send( + thread_id=str(args.get("thread_id") or ""), + message=str(args.get("message") or ""), + use_notes=bool(args.get("use_notes")), + save_lock=_write_lock, + ) + + +# Actions that stream a text/event-stream response instead of a single JSON body. +# The client (dashboard) reads them with a fetch() body reader; the AHK paste +# path never uses these. Each factory yields event dicts: +# {"type": "delta", "delta": str} one token chunk +# {"type": "done", thread_id, title, notes_used} success terminator +# {"type": "error", error, thread_id, title, notes_used} mid-stream failure +_STREAM_ACTIONS: dict[str, Callable[[dict], Iterator[dict]]] = { + "chat_send_stream": _stream_chat_send, +} + + +def _sse_frame(event: str | None, data: dict) -> bytes: + """Encode one Server-Sent-Events frame: optional ``event:`` line + JSON ``data:``.""" + prefix = f"event: {event}\n" if event else "" + return (prefix + "data: " + json.dumps(data, ensure_ascii=False) + "\n\n").encode("utf-8") + + _shutdown_event = threading.Event() _started_at = time.time() @@ -855,6 +887,46 @@ def _host_allowed(self) -> bool: host = (self.headers.get("Host") or "").split(":", 1)[0].strip().lower() return host in _ALLOWED_HOSTNAMES + def _stream_action(self, name: str, args: dict) -> None: + """Run a streaming action, forwarding its events as text/event-stream frames. + + Sends a 200 event-stream response, then flushes one SSE frame per event + from the factory so the dashboard renders tokens as they arrive. A client + disconnect (BrokenPipe) ends quietly; any other mid-stream error is sent + as a final ``event: error`` frame. Never falls through to ``_send_json`` + — the response is already committed once headers are sent.""" + factory = _STREAM_ACTIONS[name] + self.send_response(200) + self.send_header("Content-Type", "text/event-stream; charset=utf-8") + self.send_header("Cache-Control", "no-cache") + self.send_header("X-Accel-Buffering", "no") + self.send_header("X-FFP-API", API_VERSION) + self.end_headers() + start = time.time() + frames = 0 + try: + for event in factory(args): + kind = str(event.get("type") or "") + if kind == "delta": + self.wfile.write(_sse_frame(None, {"delta": event.get("delta", "")})) + elif kind in ("done", "error"): + self.wfile.write(_sse_frame(kind, {k: v for k, v in event.items() if k != "type"})) + else: + continue + self.wfile.flush() + frames += 1 + except (BrokenPipeError, ConnectionError): + log.info("action=%s stream client disconnected after %d frame(s)", name, frames) + return + except Exception as e: + log.warning("action=%s stream_error=%s", name, e) + log.debug("traceback:\n%s", traceback.format_exc()) + with contextlib.suppress(Exception): + self.wfile.write(_sse_frame("error", {"error": str(e)})) + self.wfile.flush() + elapsed = (time.time() - start) * 1000.0 + log.info("action=%s status=stream frames=%d elapsed_ms=%.1f", name, frames, elapsed) + def _send_static(self, filename: str, content_type: str) -> None: try: data = (_WEB_DIR / filename).read_bytes() @@ -909,8 +981,9 @@ def do_POST(self) -> None: # noqa: N802 self._send_json(404, {"ok": False, "error": f"POST {self.path} not found"}) return action_name = self.path[len("/action/"):].split("?", 1)[0] + stream_factory = _STREAM_ACTIONS.get(action_name) handler = ACTIONS.get(action_name) - if handler is None: + if handler is None and stream_factory is None: self._send_json(404, _err(f"unknown action: {action_name}", 0.0)) return @@ -943,6 +1016,13 @@ def do_POST(self) -> None: # noqa: N802 return args = payload.get("args") or {} + if stream_factory is not None: + # Event-stream response: commits its own headers/body and never + # returns JSON. It serializes its final write via _write_lock itself + # (passed down by the factory), so it stays outside the wrapper below. + self._stream_action(action_name, args) + return + start = time.time() try: if action_name in _WRITE_ACTIONS: diff --git a/scripts/ui/web/app.js b/scripts/ui/web/app.js index c1b5b83..60e99e0 100644 --- a/scripts/ui/web/app.js +++ b/scripts/ui/web/app.js @@ -1376,6 +1376,61 @@ async function deleteChatThread(id) { } } +// Parse one SSE frame (text between blank-line separators) into {event, data}. +// Returns null for frames without a JSON data payload. +function parseSseFrame(frame) { + let event = "message"; + const dataLines = []; + for (const line of frame.split("\n")) { + if (line.startsWith("event:")) event = line.slice(6).trim(); + else if (line.startsWith("data:")) dataLines.push(line.slice(5).replace(/^ /, "")); + } + if (!dataLines.length) return null; + try { return { event, data: JSON.parse(dataLines.join("\n")) }; } + catch (e) { return null; } +} + +// Stream chat_send_stream: call onDelta(text) per token chunk; resolve with the +// terminal metadata ({thread_id, title, notes_used, error}). Throws only when the +// stream can't be opened (old daemon / transport) so the caller can fall back to +// the one-shot chat_send action; a mid-stream failure resolves with .error set. +async function streamChat(args, onDelta) { + const res = await fetch("/action/chat_send_stream", { + method: "POST", + headers: { "Content-Type": "application/json; charset=utf-8", "X-FFP-API": API_HEADER }, + body: JSON.stringify({ args }), + }); + if (!res.ok || !res.body) throw new Error(`stream unavailable (HTTP ${res.status})`); + const reader = res.body.getReader(); + const decoder = new TextDecoder(); + const result = { thread_id: args.thread_id, title: "", notes_used: [], error: "" }; + let buf = ""; + try { + for (;;) { + const { value, done } = await reader.read(); + if (done) break; + buf += decoder.decode(value, { stream: true }); + let sep; + while ((sep = buf.indexOf("\n\n")) >= 0) { + const evt = parseSseFrame(buf.slice(0, sep)); + buf = buf.slice(sep + 2); + if (!evt) continue; + if (evt.event === "done" || evt.event === "error") { + Object.assign(result, evt.data); + if (evt.event === "error") result.error = evt.data.error || "stream failed"; + } else if (evt.data && typeof evt.data.delta === "string") { + onDelta(evt.data.delta); + } + } + } + } catch (e) { + // Reader broke after the stream opened (and the turn was already persisted + // server-side) — surface a partial, never re-send via the fallback path. + result.error = result.error || e.message || "stream interrupted"; + } + return result; +} + async function sendChat() { const input = $("chat-input"); const message = input.value.trim(); @@ -1383,7 +1438,7 @@ async function sendChat() { const btn = $("chat-send"); btn.disabled = true; setStatus("chat-status", "Thinking…"); - // Optimistically show the user's message; the reply lands when the model returns. + // Optimistically show the user's message; the reply streams in below it. const box = $("chat-transcript"); const userDiv = document.createElement("div"); userDiv.className = "chat-msg chat-msg-user"; @@ -1392,23 +1447,38 @@ async function sendChat() { $("chat-placeholder").hidden = true; box.scrollTop = box.scrollHeight; input.value = ""; + const useNotes = $("chat-use-notes").checked; + + // Assistant bubble, filled incrementally as tokens arrive. + const reply = document.createElement("div"); + reply.className = "chat-msg chat-msg-assistant"; + box.append(reply); + const showNotes = (notes) => + setStatus("chat-status", notes && notes.length ? `📚 Grounded in: ${notes.join(", ")}` : ""); + try { - const res = await action("chat_send", { - thread_id: chatThreadId, - message, - use_notes: $("chat-use-notes").checked, + const out = await streamChat({ thread_id: chatThreadId, message, use_notes: useNotes }, (delta) => { + reply.textContent += delta; + box.scrollTop = box.scrollHeight; }); - chatThreadId = res.thread_id || chatThreadId; - const reply = document.createElement("div"); - reply.className = "chat-msg chat-msg-assistant"; - reply.textContent = res.reply || "(no reply)"; - box.append(reply); - box.scrollTop = box.scrollHeight; - setStatus("chat-status", - res.notes_used && res.notes_used.length ? `📚 Grounded in: ${res.notes_used.join(", ")}` : ""); + chatThreadId = out.thread_id || chatThreadId; + if (!reply.textContent) reply.textContent = "(no reply)"; + if (out.error) setStatus("chat-status", `Send failed: ${out.error}`, false); + else showNotes(out.notes_used); loadChatThreads(); - } catch (e) { - setStatus("chat-status", `Send failed: ${e.message}`, false); + } catch (streamErr) { + // Streaming endpoint unavailable — fall back to the one-shot JSON action. + try { + const res = await action("chat_send", { thread_id: chatThreadId, message, use_notes: useNotes }); + chatThreadId = res.thread_id || chatThreadId; + reply.textContent = res.reply || "(no reply)"; + box.scrollTop = box.scrollHeight; + showNotes(res.notes_used); + loadChatThreads(); + } catch (e) { + if (!reply.textContent) reply.remove(); + setStatus("chat-status", `Send failed: ${e.message}`, false); + } } finally { btn.disabled = false; input.focus(); diff --git a/tests/test_ffp_chat_stream.py b/tests/test_ffp_chat_stream.py new file mode 100644 index 0000000..f6ff981 --- /dev/null +++ b/tests/test_ffp_chat_stream.py @@ -0,0 +1,196 @@ +"""Streaming chat: SSE delta parsing + ffp_chat.stream_send generator contract. + +Covers Phase 3 of the prompt-mode refresh — the Chat tab streams reply tokens as +they arrive. stream_send() is the daemon-side twin of send(): it yields +``{"type": "delta"}`` events, then exactly one terminal ``done``/``error`` event, +and persists the (possibly partial) turn just like the one-shot path. +""" +from __future__ import annotations + +import threading + +import ffp_chat +import pytest + + +@pytest.fixture(autouse=True) +def _tmp_store(tmp_path, monkeypatch): + monkeypatch.setattr(ffp_chat, "THREADS_PATH", tmp_path / "chat_threads.jsonl") + monkeypatch.setattr(ffp_chat, "STAGED_PATH", tmp_path / ".chat_staged.json") + return tmp_path + + +def _drive(gen): + """Split a stream_send() run into (delta strings, terminal event).""" + events = list(gen) + deltas = [e["delta"] for e in events if e.get("type") == "delta"] + terminals = [e for e in events if e.get("type") in ("done", "error")] + assert len(terminals) == 1, f"expected exactly one terminal event, got {terminals}" + return deltas, terminals[0] + + +# ---------- _parse_sse_delta ----------------------------------------------------------- + +@pytest.mark.parametrize(("line", "expected"), [ + ('data: {"choices":[{"delta":{"content":"Hi"}}]}', ["Hi"]), + (b'data: {"choices":[{"delta":{"content":"bytes"}}]}', ["bytes"]), + ("data: [DONE]", []), # terminator + ("", []), # blank separator + (": keep-alive comment", []), # SSE comment, not data + ("event: done", []), # non-data line + ('data: {"choices":[{"delta":{}}]}', []), # role-only delta, no content + ("data: not json", []), # malformed → no throw + ('data: {"choices":[{"delta":{"content":"a"}},{"delta":{"content":"b"}}]}', ["a", "b"]), +]) +def test_parse_sse_delta(line, expected): + assert list(ffp_chat._parse_sse_delta(line)) == expected + + +# ---------- stream_send happy path ----------------------------------------------------- + +def test_stream_send_accumulates_and_persists(): + seen = {} + + def fake_stream(messages): + seen["messages"] = messages + yield from ["Hel", "lo", " world"] + + deltas, terminal = _drive(ffp_chat.stream_send(message="Do a thing", llm_stream=fake_stream)) + assert deltas == ["Hel", "lo", " world"] + assert terminal["type"] == "done" + assert terminal["title"] == "Do a thing" + assert terminal["thread_id"] + # System prompt + user turn reached the model. + assert seen["messages"][0]["role"] == "system" + assert seen["messages"][-1] == {"role": "user", "content": "Do a thing"} + # Persisted with the fully joined reply. + full = ffp_chat.get_thread(terminal["thread_id"]) + assert [m["role"] for m in full["history"]] == ["user", "assistant"] + assert full["history"][1]["content"] == "Hello world" + + +def test_stream_send_continues_thread_with_window(): + _, t1 = _drive(ffp_chat.stream_send(message="first", llm_stream=lambda m: iter(["r1"]))) + tid = t1["thread_id"] + captured = {} + + def cap(messages): + captured["m"] = messages + yield "r2" + + _, t2 = _drive(ffp_chat.stream_send(thread_id=tid, message="second", llm_stream=cap)) + assert t2["thread_id"] == tid + roles = [m["role"] for m in captured["m"]] + assert roles[-3:] == ["user", "assistant", "user"] # first, r1, second + full = ffp_chat.get_thread(tid) + assert [m["content"] for m in full["history"]] == ["first", "r1", "second", "r2"] + + +def test_stream_send_notes_grounding(monkeypatch): + monkeypatch.setattr( + ffp_chat, "retrieve_notes_context", + lambda q, max_notes=4: ("NOTES CONTEXT BLOCK", ["Note One", "Note Two"]), + ) + captured = {} + + def cap(messages): + captured["m"] = messages + yield "grounded" + + _, terminal = _drive(ffp_chat.stream_send(message="what did I save", use_notes=True, llm_stream=cap)) + assert terminal["notes_used"] == ["Note One", "Note Two"] + system_msgs = [m["content"] for m in captured["m"] if m["role"] == "system"] + assert "NOTES CONTEXT BLOCK" in system_msgs + + +def test_stream_send_empty_message_rejected(): + with pytest.raises(ValueError, match="empty chat message"): + list(ffp_chat.stream_send(message=" ", llm_stream=lambda m: iter(["x"]))) + + +# ---------- stream_send failure handling ----------------------------------------------- + +def test_stream_send_partial_reply_persisted_on_error(): + def flaky(messages): + yield "par" + yield "tial" + raise RuntimeError("provider dropped") + + deltas, terminal = _drive(ffp_chat.stream_send(message="go", llm_stream=flaky)) + assert deltas == ["par", "tial"] + assert terminal["type"] == "error" + assert "provider dropped" in terminal["error"] + # The partial reply is still saved so the transcript isn't lost. + full = ffp_chat.get_thread(terminal["thread_id"]) + assert full["history"][1]["content"] == "partial" + + +def test_stream_send_error_before_any_token_saves_nothing(): + def dead(messages): + raise RuntimeError("unreachable") + yield # pragma: no cover - marks this a generator + + deltas, terminal = _drive(ffp_chat.stream_send(message="hello", llm_stream=dead)) + assert deltas == [] + assert terminal["type"] == "error" + # No reply text → no persisted turn (thread stays empty), title falls back. + assert ffp_chat.load_threads() == [] + assert terminal["title"] == "New chat" + + +def test_stream_send_uses_save_lock_around_persist(): + events = [] + + class RecordingLock: + def __enter__(self): + events.append("enter") + return self + + def __exit__(self, *a): + events.append("exit") + return False + + _drive(ffp_chat.stream_send(message="hi", llm_stream=lambda m: iter(["ok"]), save_lock=RecordingLock())) + assert events == ["enter", "exit"] # the final write was wrapped by the lock + + +def test_stream_send_real_lock_type_accepted(): + # A real threading.Lock is a valid context manager for save_lock. + _drive(ffp_chat.stream_send(message="hi", llm_stream=lambda m: iter(["ok"]), save_lock=threading.Lock())) + assert len(ffp_chat.load_threads()) == 1 + + +# ---------- concurrency: read-modify-write must not clobber a concurrent write --------- + +def test_stream_send_does_not_clobber_concurrent_write(): + # Regression guard: _persist_turn re-reads the store under the lock, so a + # write that lands DURING the (slow) stream is not overwritten by a stale + # snapshot captured before it started. The old snapshot-at-start code would + # resurrect the deleted thread here. + ffp_chat.save_threads([ + {"thread_id": "A", "title": "A", "updated_at": "2026-01-01T00:00:00", "history": []}, + {"thread_id": "B", "title": "B", "updated_at": "2026-01-01T00:00:00", "history": []}, + ]) + + def stream(messages): + yield "hel" + ffp_chat.delete_thread("B") # a concurrent writer deletes B mid-stream + yield "lo" + + list(ffp_chat.stream_send(thread_id="A", message="hi", llm_stream=stream)) + ids = {t["thread_id"] for t in ffp_chat.load_threads()} + assert "B" not in ids # the delete survived — no stale-snapshot resurrection + a = ffp_chat.get_thread("A") + assert [m["content"] for m in a["history"]] == ["hi", "hello"] + + +def test_stream_send_persists_partial_on_client_disconnect(): + # gen.close() throws GeneratorExit at the suspended yield — exactly what the + # daemon does when it stops iterating after a client disconnect. The partial + # reply streamed so far must still be persisted. + gen = ffp_chat.stream_send(message="question?", llm_stream=lambda m: iter(["par", "tial"])) + assert next(gen)["delta"] == "par" # first token delivered, then the client drops + gen.close() + threads = ffp_chat.load_threads() + assert len(threads) == 1 + assert [m["content"] for m in threads[0]["history"]] == ["question?", "par"] diff --git a/tests/test_ffp_daemon.py b/tests/test_ffp_daemon.py index c88582b..4db0914 100644 --- a/tests/test_ffp_daemon.py +++ b/tests/test_ffp_daemon.py @@ -123,6 +123,140 @@ def test_actions_count_and_expected_names(daemon_module): assert "model_stats" not in daemon_module.ACTIONS +# ---------- streaming chat (Phase 3: SSE) ---------------------------------------------- + +def test_stream_actions_registered_separately(daemon_module): + # chat_send_stream streams text/event-stream, so it lives in _STREAM_ACTIONS, + # NOT ACTIONS (keeping the ACTIONS count/JSON contract unchanged) and NOT + # _WRITE_ACTIONS (it serializes its own final write via the passed lock). + assert "chat_send_stream" in daemon_module._STREAM_ACTIONS + assert "chat_send_stream" not in daemon_module.ACTIONS + assert "chat_send_stream" not in daemon_module._WRITE_ACTIONS + # The non-streaming one-shot action stays registered for the fallback path. + assert "chat_send" in daemon_module.ACTIONS + + +def test_sse_frame_format(daemon_module): + # Delta frame: no event line, just a JSON data line + blank separator. + assert daemon_module._sse_frame(None, {"delta": "hi"}) == b'data: {"delta": "hi"}\n\n' + # Named event frame: event line then data line. + frame = daemon_module._sse_frame("done", {"thread_id": "t1"}).decode("utf-8") + assert frame.startswith("event: done\n") + assert 'data: {"thread_id": "t1"}' in frame + assert frame.endswith("\n\n") + + +def test_chat_send_stream_end_to_end(daemon_server, monkeypatch): + daemon_module, base_url = daemon_server + import ffp_chat + + captured = {} + + def fake_stream_send(thread_id="", message="", use_notes=False, save_lock=None): + captured["args"] = (thread_id, message, use_notes) + captured["lock"] = save_lock + yield {"type": "delta", "delta": "Hel"} + yield {"type": "delta", "delta": "lo"} + yield {"type": "done", "thread_id": "tid-1", "title": "greet", "notes_used": []} + + monkeypatch.setattr(ffp_chat, "stream_send", fake_stream_send) + + body = json.dumps({"args": {"thread_id": "", "message": "hi there", "use_notes": False}}).encode("utf-8") + req = urllib.request.Request( + f"{base_url}/action/chat_send_stream", + data=body, + headers={"X-FFP-API": "1", "Content-Type": "application/json"}, + method="POST", + ) + with urllib.request.urlopen(req, timeout=5) as resp: + assert resp.status == 200 + assert resp.headers.get("Content-Type", "").startswith("text/event-stream") + raw = resp.read().decode("utf-8") + + # The daemon passed the streaming request through and used its write-lock. + assert captured["args"] == ("", "hi there", False) + assert captured["lock"] is daemon_module._write_lock + # Two delta frames then a done frame, in order. + assert '{"delta": "Hel"}' in raw + assert '{"delta": "lo"}' in raw + assert "event: done" in raw + assert raw.index('"Hel"') < raw.index('"lo"') < raw.index("event: done") + + +def test_chat_send_stream_reports_midstream_error(daemon_server, monkeypatch): + daemon_module, base_url = daemon_server + import ffp_chat + + def failing_stream_send(thread_id="", message="", use_notes=False, save_lock=None): + yield {"type": "delta", "delta": "part"} + raise RuntimeError("provider blew up") + + monkeypatch.setattr(ffp_chat, "stream_send", failing_stream_send) + + body = json.dumps({"args": {"message": "hi"}}).encode("utf-8") + req = urllib.request.Request( + f"{base_url}/action/chat_send_stream", + data=body, + headers={"X-FFP-API": "1", "Content-Type": "application/json"}, + method="POST", + ) + with urllib.request.urlopen(req, timeout=5) as resp: + raw = resp.read().decode("utf-8") + # Partial token delivered, then an error frame carrying the message. + assert '{"delta": "part"}' in raw + assert "event: error" in raw + assert "provider blew up" in raw + + +def test_stream_action_survives_client_disconnect_and_persists_partial(daemon_module, monkeypatch, tmp_path): + # When the client disconnects, wfile.write raises BrokenPipeError; _stream_action + # must swallow it and return, and abandoning the loop must finalize the real + # stream_send generator (GeneratorExit) so the partial reply is still persisted. + import ffp_chat + monkeypatch.setattr(ffp_chat, "THREADS_PATH", tmp_path / "chat_threads.jsonl") + monkeypatch.setattr(ffp_chat, "_default_llm_stream", lambda messages: iter(["par", "tial"])) + + class FakeWfile: + def __init__(self): + self.writes = 0 + + def write(self, data): + self.writes += 1 + if self.writes >= 2: # first frame lands, then the client is gone + raise BrokenPipeError("client disconnected") + + def flush(self): + pass + + handler = daemon_module.Handler.__new__(daemon_module.Handler) + handler.send_response = lambda *a, **k: None + handler.send_header = lambda *a, **k: None + handler.end_headers = lambda: None + handler.wfile = FakeWfile() + + # Must not raise despite the broken pipe. + handler._stream_action("chat_send_stream", {"message": "hi there"}) + + threads = ffp_chat.load_threads() + assert len(threads) == 1 + assert threads[0]["history"][0]["content"] == "hi there" + assert threads[0]["history"][1]["content"] == "partial" # both streamed chunks saved + + +def test_chat_send_stream_requires_api_header(daemon_server): + daemon_module, base_url = daemon_server + body = json.dumps({"args": {"message": "hi"}}).encode("utf-8") + req = urllib.request.Request( + f"{base_url}/action/chat_send_stream", + data=body, + headers={"Content-Type": "application/json"}, # missing X-FFP-API + method="POST", + ) + with pytest.raises(urllib.error.HTTPError) as exc: + urllib.request.urlopen(req, timeout=5) + assert exc.value.code == 403 + + def test_v31_model_warm_scheduler_runs_startup_and_idle_interval(daemon_module): cfg = { "llm": {"provider": "fastflowlm"}, From 3bf307166976f200ec3bc1b5a6976af643f274bd Mon Sep 17 00:00:00 2001 From: agrechenkov Date: Mon, 27 Jul 2026 16:42:13 -0400 Subject: [PATCH 2/2] 2.4.0: streaming Chat tab Bump version across _version.py, pyproject.toml, installer.iss, README (+ What's new in 2.4), sign.ps1 example, and the version-sync test pin. Move CHANGELOG Unreleased -> 2.4.0. Co-Authored-By: Claude Opus 4.8 --- CHANGELOG.md | 10 +++++++++- README.md | 6 +++++- installer/installer.iss | 2 +- installer/sign.ps1 | 2 +- pyproject.toml | 2 +- scripts/_version.py | 2 +- tests/test_version_sync.py | 2 +- 7 files changed, 19 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0ea36e0..d7d03a7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,9 +2,17 @@ ## Unreleased +## 2.4.0 + +**Streaming chat.** The dashboard Chat tab renders replies token-by-token instead of waiting for the whole answer, so the response starts appearing at warm time-to-first-token rather than after the full completion. + ### Added -- **The Chat tab streams replies as they generate.** Instead of a blank "Thinking…" wait until the whole answer lands, the assistant's reply now fills in token-by-token over a Server-Sent-Events stream — first text typically appears at ~1.6 s (warm TTFT) rather than after the full completion. Prompt-mode/grammar hotkeys and the AHK paste path are unchanged (they still return whole output). A daemon without the streaming endpoint, or any failure to open the stream, transparently falls back to the previous one-shot request, so nothing regresses. Works on both FastFlowLM and Ollama (shared OpenAI-compatible SSE). A mid-stream provider drop still saves the partial reply and shows the error. +- **The Chat tab streams replies as they generate.** Instead of a blank "Thinking…" wait until the whole answer lands, the assistant's reply now fills in token-by-token over a Server-Sent-Events stream — first text typically appears at ~1.6 s (warm TTFT) rather than after the full completion. Measured live on FastFlowLM `qwen3.5:4b`: first token at **1.52–1.58 s** versus a ~2 s (short) to tens-of-seconds (long) full-completion wait. Prompt-mode/grammar hotkeys and the AHK paste path are unchanged (they still return whole output). A daemon without the streaming endpoint, or any failure to open the stream, transparently falls back to the previous one-shot request, so nothing regresses. Works on both FastFlowLM and Ollama (shared OpenAI-compatible SSE). A mid-stream provider drop or a client disconnect still saves the partial reply. + +### Internal + +- Streaming persistence re-reads the thread store under the daemon write-lock (atomic read-modify-write), so a concurrent chat write or thread delete during a multi-second stream is never clobbered by a stale snapshot; a client disconnect (`GeneratorExit`) still persists the partial turn. Both were caught by an adversarial review pass and pinned by discriminating tests (SPEC V37 / T27). ## 2.3.0 diff --git a/README.md b/README.md index facec65..657b254 100644 --- a/README.md +++ b/README.md @@ -4,7 +4,11 @@ Flowkey is a Windows desktop assistant that adds local-LLM hotkeys for grammar f Everything runs locally through [FastFlowLM](https://fastflowlm.com) (AMD Ryzen AI NPU) or, on machines without the NPU, through [Ollama](https://ollama.com) (CPU/GPU) as a secondary provider. No cloud service, analytics, or telemetry is used by the app. -Current version: `2.3.0` +Current version: `2.4.0` + +## What's new in 2.4 + +- **Chat streams as it thinks.** The dashboard **Chat** tab now fills the reply in token-by-token over a live stream instead of waiting on a blank "Thinking…" until the whole answer is ready — the first words typically appear in ~1.6 s (warm) rather than after the full response. It works on both FastFlowLM and Ollama, and if anything interrupts the stream the partial reply is kept and the error is shown. Grammar, `prompt:`, and the other hotkeys are unchanged (they still paste the finished result). ## What's new in 2.3 diff --git a/installer/installer.iss b/installer/installer.iss index 5191c86..cd2e88b 100644 --- a/installer/installer.iss +++ b/installer/installer.iss @@ -41,7 +41,7 @@ #define AppURL "https://github.com/agr77one/Fastflow" #define AppExeName "Flowkey.exe" ; symbolic — actual launchers below ; Keep in lockstep with scripts\_version.py. -#define AppVersion "2.3.0" +#define AppVersion "2.4.0" [Setup] AppId={{8A4F1E6C-9B3D-4E62-9F7A-FASTFLOW140}} diff --git a/installer/sign.ps1 b/installer/sign.ps1 index baed345..6248553 100644 --- a/installer/sign.ps1 +++ b/installer/sign.ps1 @@ -52,7 +52,7 @@ .EXAMPLE # Sign the installer $env:FFP_SIGN_PFX_PASSWORD = "ChangeMe!" - .\sign.ps1 -FilePath ..\out\Flowkey-Setup-2.3.0.exe + .\sign.ps1 -FilePath ..\out\Flowkey-Setup-2.4.0.exe #> [CmdletBinding(DefaultParameterSetName = "Sign")] diff --git a/pyproject.toml b/pyproject.toml index 60f4149..7d7682a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -17,7 +17,7 @@ build-backend = "setuptools.build_meta" [project] name = "fastflowprompt" -version = "2.3.0" +version = "2.4.0" description = "Local-LLM-powered grammar fix, prompt rewrite, chat, and dashboard for Windows." readme = "README.md" requires-python = ">=3.11" diff --git a/scripts/_version.py b/scripts/_version.py index fdb5750..9458797 100644 --- a/scripts/_version.py +++ b/scripts/_version.py @@ -1,3 +1,3 @@ """Single source of truth for the app version. Read by grammar_fix.py.""" -__version__ = "2.3.0" +__version__ = "2.4.0" diff --git a/tests/test_version_sync.py b/tests/test_version_sync.py index 4e071a0..a291823 100644 --- a/tests/test_version_sync.py +++ b/tests/test_version_sync.py @@ -32,4 +32,4 @@ def test_v18_release_version_is_synchronized(): ), } - assert set(versions.values()) == {"2.3.0"}, versions + assert set(versions.values()) == {"2.4.0"}, versions