From 6bc98e2b62ce883e591a94ff234309ec00f07f7f Mon Sep 17 00:00:00 2001 From: mARTin-B78 Date: Sun, 5 Jul 2026 18:08:54 +0200 Subject: [PATCH] Fix ASGI middleware crash and stream/blocking lock race (v1.12.96) The static-asset caching middleware used BaseHTTPMiddleware, which has a known Starlette bug: a client disconnecting mid-StreamingResponse (the new live-attribution SSE stream hitting its idle timeout) raced its internal task group and raised "RuntimeError: No response returned", crashing that request. Rewritten as plain ASGI middleware that only touches headers via the raw send callable, removing the race. Also found the real cause of the casting timeouts/405s: the streaming attribution endpoint had its own lock instead of sharing the one the blocking endpoint already used to serialize on the LLM's single slot - letting a stream call and its own blocking fallback fire concurrently, exactly the ghost-request pile-up that lock was built to prevent. Unified onto one lock and added server-side logging for stream failures. The "LLM Thinking" pane now shows the model's actual reasoning instead of the in-progress JSON answer echoed back at the user. Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 11 ++++++++ VERSION | 2 +- routes/conversation.py | 56 +++++++++++++++++++++++++++----------- server.py | 61 +++++++++++++++++++++++++++++------------- static/index.html | 6 ++--- static/js/audiobook.js | 15 ++++++++--- 6 files changed, 109 insertions(+), 42 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c9eaa5f..b7cb52a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,17 @@ Follows [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) · versioned wi --- +## [1.12.96] — 2026-07-05 + +### Fixed +- **Server crashes with "RuntimeError: No response returned"** — the static-asset caching middleware used `@app.middleware("http")` (Starlette's `BaseHTTPMiddleware`), which has a known bug: when a client disconnects mid-`StreamingResponse` (e.g. the new live-attribution SSE stream hitting its idle timeout), its internal task group races the disconnect and raises this error. Rewritten as plain ASGI middleware that only touches response headers via the raw `send` callable — it never wraps the response the way `call_next()` does, so the race is gone entirely. +- **Casting timeouts/405s during long runs** — the streaming attribution endpoint used its own lock, separate from the blocking endpoint's. That meant a stream request and its own blocking fallback could both fire into the LLM's single processing slot at once — exactly the "ghost request" pile-up the original lock existed to prevent. Both endpoints now share one lock. Stream failures are also logged server-side now (silent before). + +### Changed +- **The "LLM Thinking…" pane now shows real reasoning**, not the JSON answer echoed back. The attribution prompt asks for a brief `...` rationale before the JSON when streaming; the client shows only that block and stops once it closes, instead of dumping the raw in-progress JSON (which mostly just reproduces the passage text). + +--- + ## [1.12.95] — 2026-07-05 ### Added diff --git a/VERSION b/VERSION index a65ccb9..2fbc2df 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -1.12.95 +1.12.96 diff --git a/routes/conversation.py b/routes/conversation.py index 4faf0ef..4fdb02b 100644 --- a/routes/conversation.py +++ b/routes/conversation.py @@ -40,7 +40,14 @@ router = APIRouter() # occupying the LLM's single processing slot. Without this lock, the next # chunk (or the retry-in-halves) fires into that busy slot and the LLM # answers with 429s that cascade until the ghosts drain. -_attribution_llm_lock = asyncio.Lock() +# threading.Lock (not asyncio.Lock) because the streaming endpoint's SSE +# generator runs in a plain thread (StreamingResponse iterates a sync +# generator via a threadpool), so both the blocking and streaming endpoints +# must share ONE lock object to actually serialize on the LLM's single slot — +# two separate locks would let a stream call and its own blocking fallback +# fire into that slot concurrently, which is exactly the "ghost request" +# scenario this lock exists to prevent. +_attribution_llm_lock = threading.Lock() # ── STT hallucination filter ────────────────────────────────────────────────── # Whisper commonly hallucinates these phrases on silence/noise. @@ -1020,12 +1027,25 @@ def _attribution_prepare(data: dict) -> dict: if lang_hint: base_prompt += f"\n\n{lang_hint}" - system = ( - f"{base_prompt}\n\n" - "You MUST respond with a single, valid JSON object containing a 'segments' array.\n" - "Do NOT output any reasoning, chain of thought, or conversational text. Output ONLY the raw JSON object:\n" - '{"segments":[{"speaker":"Narrator","type":"narration","text":"...","emotion":""}]}' - ) + # Only the streaming ("watch it think") call asks for a reasoning preamble — + # it costs extra tokens/latency, which the blocking/fallback path can't + # afford when it's already the retry-in-halves path for a slow model. + if data.get("want_reasoning"): + system = ( + f"{base_prompt}\n\n" + "First, in a ... block, briefly reason (2-4 short sentences) about who speaks each " + "untagged or ambiguous quote in this passage — mention the names/pronouns you're resolving. Keep it short.\n" + "Then, after , respond with a single, valid JSON object containing a 'segments' array. " + "Nothing else outside the block and the JSON object:\n" + '{"segments":[{"speaker":"Narrator","type":"narration","text":"...","emotion":""}]}' + ) + else: + system = ( + f"{base_prompt}\n\n" + "You MUST respond with a single, valid JSON object containing a 'segments' array.\n" + "Do NOT output any reasoning, chain of thought, or conversational text. Output ONLY the raw JSON object:\n" + '{"segments":[{"speaker":"Narrator","type":"narration","text":"...","emotion":""}]}' + ) user = ( ("Known characters so far: " + ", ".join(str(n) for n in known) + "\n\n" if known else "") + ("Recent dialogue (the immediately preceding lines — continue the same conversation/turn-taking):\n" + recent + "\n\n" if recent else "") @@ -1044,7 +1064,7 @@ def _attribution_prepare(data: dict) -> dict: # tighter cap here once did) truncates mid-JSON on exchange-heavy scenes # — which the parser can't always repair, degrading the whole chunk to # naive quote-splitting with every speaker labelled "Unknown". - "max_tokens": min(8192, max(2048, len(text) + 1000)), + "max_tokens": min(8192, max(2048, len(text) + 1000)) + (250 if data.get("want_reasoning") else 0), } if model: payload["model"] = model @@ -1071,7 +1091,9 @@ async def attribute_dialogue(request: Request): timeout_seconds, text = prep["timeout_seconds"], prep["text"] api_key = prep["api_key"] try: - async with _attribution_llm_lock: + if not await asyncio.to_thread(_attribution_llm_lock.acquire, True, timeout_seconds): + return _fallback_attribute_response(text) + try: resp = await asyncio.to_thread( _post_llm_chat_completion, llm_url, payload, @@ -1093,6 +1115,8 @@ async def attribute_dialogue(request: Request): {"Authorization": f"Bearer {api_key}"}, timeout_seconds, ) payload = retry_payload + finally: + _attribution_llm_lock.release() if resp.status_code >= 400: print( f"[attribute-dialogue] LLM failed ({resp.status_code}) for model " @@ -1157,11 +1181,6 @@ def _attribution_parse(raw: str, text: str) -> dict: return {"segments": clean, "characters": chars} -# Streaming attribution runs in a plain thread (StreamingResponse with a sync -# generator), so it serializes on a threading.Lock rather than the asyncio one. -_attribution_stream_tlock = threading.Lock() - - @router.post("/api/attribute-dialogue/stream") async def attribute_dialogue_stream(request: Request): """Same job as /api/attribute-dialogue, but streams the LLM's live output @@ -1176,7 +1195,7 @@ async def attribute_dialogue_stream(request: Request): headers = {"Authorization": f"Bearer {prep['api_key']}"} def gen(): - if not _attribution_stream_tlock.acquire(timeout=prep["timeout_seconds"]): + if not _attribution_llm_lock.acquire(timeout=prep["timeout_seconds"]): yield 'data: {"error": "Attribution engine busy"}\n\n' return upstream = None @@ -1198,9 +1217,14 @@ async def attribute_dialogue_stream(request: Request): last_err = exc upstream = None if upstream is None: + print(f"[attribute-dialogue/stream] LLM connect failed: {last_err}") yield f'data: {json.dumps({"error": str(last_err or "LLM connect failed")})}\n\n' return if upstream.status_code >= 400: + print( + f"[attribute-dialogue/stream] LLM failed ({upstream.status_code}) for model " + f"{payload.get('model') or '(default)'}: {_response_error_text(upstream)[:300]}" + ) yield f'data: {json.dumps({"error": f"HTTP {upstream.status_code}: {_response_error_text(upstream)[:300]}"})}\n\n' return raw_parts = [] @@ -1234,7 +1258,7 @@ async def attribute_dialogue_stream(request: Request): upstream.close() except Exception: pass - _attribution_stream_tlock.release() + _attribution_llm_lock.release() return StreamingResponse(gen(), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}) diff --git a/server.py b/server.py index 59c367d..93baf11 100644 --- a/server.py +++ b/server.py @@ -7,10 +7,12 @@ from logging.handlers import RotatingFileHandler from pathlib import Path from contextlib import asynccontextmanager -from fastapi import FastAPI, Request -from fastapi.responses import Response +from urllib.parse import parse_qs + +from fastapi import FastAPI from fastapi.staticfiles import StaticFiles from fastapi.middleware.gzip import GZipMiddleware +from starlette.datastructures import MutableHeaders from core.constants import STATIC_DIR, _BufferHandler from core.config import _load_settings @@ -73,24 +75,45 @@ except Exception as _log_err: # ── Static-asset caching middleware ────────────────────────────────────────── # JS and CSS are served with ?v= by loader.js — safe to cache for 1 year. # Section HTML uses ?v= (always fresh) so no long-term cache there. +# +# Pure ASGI middleware, not @app.middleware("http") (= Starlette's +# BaseHTTPMiddleware) — that implementation wraps every request in a task +# group around call_next(), and a client disconnecting mid-StreamingResponse +# (e.g. the live-attribution SSE stream hitting its idle timeout) races that +# task group and raises "RuntimeError: No response returned." This class +# only touches response headers via the raw `send` callable, so it never +# holds the response open the way call_next() does. +class StaticCacheHeadersMiddleware: + def __init__(self, app): + self.app = app -@app.middleware("http") -async def static_cache_headers(request: Request, call_next): - response: Response = await call_next(request) - path = request.url.path - has_version = bool(request.query_params.get("v")) - if has_version and (path.startswith("/static/js/") or path.endswith(".css")): - response.headers["Cache-Control"] = "public, max-age=31536000, immutable" - elif path.startswith("/static/sections/"): - response.headers["Cache-Control"] = "no-store" - elif path.startswith("/static/vendor/"): - response.headers["Cache-Control"] = "public, max-age=86400" - elif path.startswith("/static/") and not has_version: - # Bootstrap files (style.css, loader.js, nav.js, …) are loaded without a ?v= query. - # Revalidate on every load so GUI edits appear after a normal reload — the server - # returns 304 when the file is unchanged, so this stays cheap. - response.headers["Cache-Control"] = "no-cache" - return response + async def __call__(self, scope, receive, send): + if scope["type"] != "http": + await self.app(scope, receive, send) + return + path = scope["path"] + has_version = "v" in parse_qs((scope.get("query_string") or b"").decode()) + + async def send_wrapper(message): + if message["type"] == "http.response.start": + headers = MutableHeaders(scope=message) + if has_version and (path.startswith("/static/js/") or path.endswith(".css")): + headers["Cache-Control"] = "public, max-age=31536000, immutable" + elif path.startswith("/static/sections/"): + headers["Cache-Control"] = "no-store" + elif path.startswith("/static/vendor/"): + headers["Cache-Control"] = "public, max-age=86400" + elif path.startswith("/static/") and not has_version: + # Bootstrap files (style.css, loader.js, nav.js, …) are loaded without a ?v= query. + # Revalidate on every load so GUI edits appear after a normal reload — the server + # returns 304 when the file is unchanged, so this stays cheap. + headers["Cache-Control"] = "no-cache" + await send(message) + + await self.app(scope, receive, send_wrapper) + + +app.add_middleware(StaticCacheHeadersMiddleware) # ── Routers ─────────────────────────────────────────────────────────────────── diff --git a/static/index.html b/static/index.html index 2a20f81..ce34f1d 100644 --- a/static/index.html +++ b/static/index.html @@ -10,7 +10,7 @@ - + @@ -27,7 +27,7 @@ - + @@ -365,7 +365,7 @@ window.toggleNavTree = function(treeId, chevronId) { - + diff --git a/static/js/audiobook.js b/static/js/audiobook.js index f3050c1..d4a758d 100644 --- a/static/js/audiobook.js +++ b/static/js/audiobook.js @@ -82,7 +82,7 @@ async function audiobookAttributeStream(body, view, outerSignal, idleTimeoutMs = armIdle(); const r = await fetch('/api/attribute-dialogue/stream', { method: 'POST', headers: { 'Content-Type': 'application/json' }, - signal: ctl.signal, body: JSON.stringify(body), + signal: ctl.signal, body: JSON.stringify({ ...body, want_reasoning: true }), }); if (!r.ok || !r.body) throw new Error('stream HTTP ' + r.status); const reader = r.body.getReader(); @@ -2782,6 +2782,8 @@ STRIKTE FORMAT- UND TEXTREGELN: }, processing(text) { if (this._procRow) this._procRow.remove(); + this._thinkRaw = ''; + this._thinkDone = false; this._procRow = document.createElement('div'); this._procRow.className = 'ab-cv-row is-processing ab-cv-llm-row'; const isLong = text.length > 160; @@ -2821,7 +2823,7 @@ STRIKTE FORMAT- UND TEXTREGELN: // the plain preview for the split thinking/passage view; subsequent deltas // append and keep the thinking pane scrolled to the newest text. thinking(delta) { - if (!this._procRow || !delta) return; + if (!this._procRow || !delta || this._thinkDone) return; const split = this._procRow.querySelector('.ab-cv-llm-split'); const think = this._procRow.querySelector('.ab-cv-think'); if (!split || !think) return; @@ -2830,8 +2832,15 @@ STRIKTE FORMAT- UND TEXTREGELN: const prev = this._procRow.querySelector('.ab-cv-llm-preview'); if (prev) prev.hidden = true; } - think.textContent = (think.textContent + delta).slice(-20000); // cap runaway output + // The model wraps its reasoning in ...; once it closes, + // stop appending — everything after is the raw JSON answer, not + // "thinking", and dumping it here just reproduces the passage text. + this._thinkRaw = (this._thinkRaw || '') + delta; + const closeAt = this._thinkRaw.indexOf(''); + const shown = closeAt >= 0 ? this._thinkRaw.slice(0, closeAt) : this._thinkRaw; + think.textContent = shown.replace(//g, '').trim().slice(-20000); think.scrollTop = think.scrollHeight; + if (closeAt >= 0) this._thinkDone = true; }, clearProcessing() { if (this._procRow) { this._procRow.remove(); this._procRow = null; }