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 <think> reasoning instead of the in-progress JSON answer echoed back at the user. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-05 18:08:54 +02:00 · 2026-07-05 18:08:54 +02:00 · 6bc98e2b62
commit 6bc98e2b62
parent 3503f5bda6
6 changed files with 109 additions and 42 deletions
--- 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 `<think>...</think>` 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
--- a/2
+++ b/2
@ -1 +1 @@
-1.12.95
+1.12.96
--- 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,6 +1027,19 @@ def _attribution_prepare(data: dict) -> dict:
        if lang_hint:
            base_prompt += f"\n\n{lang_hint}"

+    # 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 <think>...</think> 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 </think>, respond with a single, valid JSON object containing a 'segments' array. "
+            "Nothing else outside the <think> 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"
@ -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"})
--- 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=<version> by loader.js — safe to cache for 1 year.
 # Section HTML uses ?v=<timestamp> (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"))
+    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")):
-        response.headers["Cache-Control"] = "public, max-age=31536000, immutable"
+                    headers["Cache-Control"] = "public, max-age=31536000, immutable"
                elif path.startswith("/static/sections/"):
-        response.headers["Cache-Control"] = "no-store"
+                    headers["Cache-Control"] = "no-store"
                elif path.startswith("/static/vendor/"):
-        response.headers["Cache-Control"] = "public, max-age=86400"
+                    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
+                    headers["Cache-Control"] = "no-cache"
+            await send(message)
+
+        await self.app(scope, receive, send_wrapper)
+
+
+app.add_middleware(StaticCacheHeadersMiddleware)


 # ── Routers ───────────────────────────────────────────────────────────────────
--- a/static/index.html
+++ b/static/index.html
@ -10,7 +10,7 @@
  <meta name="format-detection" content="telephone=no">
  <meta name="color-scheme" content="light dark">
  <meta name="theme-color" content="#2563EB">
-  <meta name="app-version" content="1.12.95">
+  <meta name="app-version" content="1.12.96">
  <link rel="manifest" href="/manifest.webmanifest">
  <link rel="icon" href="/static/icon.svg" type="image/svg+xml">
  <link rel="apple-touch-icon" href="/static/icon.svg">
@ -27,7 +27,7 @@

  <!-- ── Core styles (local — no CDN dependency for first paint) ────────── -->
  <link rel="stylesheet" href="/static/vendor/mdi/materialdesignicons.min.css">
-  <link rel="stylesheet" href="/static/style.css?v=1.12.95">
+  <link rel="stylesheet" href="/static/style.css?v=1.12.96">


  <!-- ── Flag icons — non-blocking (loaded async, icons appear after JS) ── -->
@ -365,7 +365,7 @@ window.toggleNavTree = function(treeId, chevronId) {
 </script>

 <!-- loader.js: fetches sections → loads JS modules → removes skeleton -->
-<script src="/static/loader.js?v=1.12.95"></script>
+<script src="/static/loader.js?v=1.12.96"></script>

 </body>
 </html>
--- 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 <think>...</think>; 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('</think>');
+      const shown = closeAt >= 0 ? this._thinkRaw.slice(0, closeAt) : this._thinkRaw;
+      think.textContent = shown.replace(/<think>/g, '').trim().slice(-20000);
      think.scrollTop = think.scrollHeight;
+      if (closeAt >= 0) this._thinkDone = true;
    },
    clearProcessing() {
      if (this._procRow) { this._procRow.remove(); this._procRow = null; }
 @ -1 +1 @@
 .12.95
 .12.96