fix(chat): log sanitized client sse diagnostics

(cherry picked from commit 749ca6e18c5e307fbf7e7fb5fffce97249545017)

CHANGELOG.md

@@ -5,7 +5,7 @@
 
 ### Fixed
 
-- Clarify `Response interrupted` recovery markers so they report that the live response stream stopped instead of asserting that the WebUI process restarted. The recovery path now records distinct interruption causes for real process restarts, stream/run split-brain, and lost worker bookkeeping; browser-side SSE transport failures show a separate `Connection interrupted` message, and client-side `BrokenPipeError` disconnects no longer get logged as server 500s.
+- Clarify `Response interrupted` recovery markers so they report that the live response stream stopped instead of asserting that the WebUI process restarted. The recovery path now records distinct interruption causes for real process restarts, stream/run split-brain, and lost worker bookkeeping; browser-side SSE transport failures show a separate `Connection interrupted` message, client-side `BrokenPipeError` disconnects no longer get logged as server 500s, and chat/gateway SSE errors emit rate-limited, body-capped, sanitized client diagnostics to `/api/client-events/log` for future root-cause checks. The stream-status `terminal_state` value for lost-worker bookkeeping changes from `stale-from-restart` to `lost-worker-bookkeeping`, matching the new non-restart wording.
 
 ## [v0.51.131] — 2026-05-24 — Release DC (stage-batch13 — 6-PR notes-drawer + context-parity + PWA-swipe + locale polish)
 

api/routes.py

@@ -23,7 +23,7 @@ import uuid
 import re
 from pathlib import Path
 from contextlib import closing
-from urllib.parse import parse_qs
+from urllib.parse import parse_qs, urlsplit
 from api.agent_sessions import (
     MESSAGING_SOURCES,
     is_cli_session_row,
@@ -75,6 +75,21 @@ _CSP_REPORT_RATE_LIMIT_LOCK = threading.Lock()
 _CSP_REPORT_RATE_LIMIT_WINDOW_SECONDS = 60
 _CSP_REPORT_RATE_LIMIT_MAX = 100
 _CSP_REPORT_MAX_BODY_BYTES = 64 * 1024
+_CLIENT_EVENT_LOGGER = logging.getLogger("client_event")
+_CLIENT_EVENT_RATE_LIMIT: dict[str, list[float]] = {}
+_CLIENT_EVENT_RATE_LIMIT_LOCK = threading.Lock()
+_CLIENT_EVENT_RATE_LIMIT_WINDOW_SECONDS = 60
+_CLIENT_EVENT_RATE_LIMIT_MAX = 30
+_CLIENT_EVENT_MAX_BODY_BYTES = 4 * 1024
+_CLIENT_EVENT_ALLOWED_FIELDS = {
+    "event": 64,
+    "source": 80,
+    "session_id": 128,
+    "stream_id": 128,
+    "visibility_state": 32,
+    "url_path": 256,
+    "reason": 160,
+}
 
 
 def _session_field(session, field, default=None):
@@ -1371,6 +1386,20 @@ def _csp_report_rate_limited(handler, *, now: float | None = None) -> bool:
     return False
 
 
+def _client_event_rate_limited(handler, *, now: float | None = None) -> bool:
+    now = time.time() if now is None else now
+    key = _client_ip_for_rate_limit(handler)
+    cutoff = now - _CLIENT_EVENT_RATE_LIMIT_WINDOW_SECONDS
+    with _CLIENT_EVENT_RATE_LIMIT_LOCK:
+        timestamps = [ts for ts in _CLIENT_EVENT_RATE_LIMIT.get(key, []) if ts >= cutoff]
+        if len(timestamps) >= _CLIENT_EVENT_RATE_LIMIT_MAX:
+            _CLIENT_EVENT_RATE_LIMIT[key] = timestamps
+            return True
+        timestamps.append(now)
+        _CLIENT_EVENT_RATE_LIMIT[key] = timestamps
+    return False
+
+
 def _send_no_content(handler, status: int = 204) -> bool:
     handler.send_response(status)
     handler.send_header("Content-Length", "0")
@@ -1410,6 +1439,105 @@ def _handle_csp_report(handler) -> bool:
     return _send_no_content(handler)
 
 
+def _bounded_client_event_string(value, limit: int) -> str | None:
+    if value is None:
+        return None
+    text = str(value).strip()
+    if not text:
+        return None
+    return text[:limit]
+
+
+def _sanitize_client_event_url_path(value) -> str | None:
+    text = _bounded_client_event_string(value, 1024)
+    if not text:
+        return None
+    try:
+        parsed = urlsplit(text)
+        path = parsed.path or "/"
+    except Exception:
+        path = text.split("?", 1)[0] or "/"
+    if not path.startswith("/"):
+        path = "/" + path.lstrip("/")
+    return path[: _CLIENT_EVENT_ALLOWED_FIELDS["url_path"]]
+
+
+def _sanitize_client_event_payload(payload: dict | None) -> dict:
+    """Whitelist tiny browser diagnostic events and discard sensitive content.
+
+    Client-side SSE diagnostics should explain transport failures without
+    persisting prompts, cookies, query strings, headers, or arbitrary browser
+    payloads. This helper intentionally keeps only bounded scalar metadata.
+    """
+    if not isinstance(payload, dict):
+        return {"event": "unknown"}
+    sanitized: dict[str, object] = {}
+    for field, limit in _CLIENT_EVENT_ALLOWED_FIELDS.items():
+        if field == "url_path":
+            value = _sanitize_client_event_url_path(payload.get(field))
+        else:
+            value = _bounded_client_event_string(payload.get(field), limit)
+        if value is not None:
+            sanitized[field] = value
+    ready_state = payload.get("ready_state")
+    if isinstance(ready_state, bool):
+        pass
+    elif isinstance(ready_state, int) and 0 <= ready_state <= 3:
+        sanitized["ready_state"] = ready_state
+    online = payload.get("online")
+    if isinstance(online, bool):
+        sanitized["online"] = online
+    elif isinstance(online, str):
+        lowered = online.strip().lower()
+        if lowered in {"true", "1", "yes", "on"}:
+            sanitized["online"] = True
+        elif lowered in {"false", "0", "no", "off"}:
+            sanitized["online"] = False
+    if "event" not in sanitized:
+        sanitized["event"] = "unknown"
+    return sanitized
+
+
+def _read_client_event_payload(handler) -> dict:
+    try:
+        length = int(handler.headers.get("Content-Length", 0))
+    except Exception:
+        length = 0
+    if length > _CLIENT_EVENT_MAX_BODY_BYTES:
+        try:
+            handler.rfile.read(_CLIENT_EVENT_MAX_BODY_BYTES)
+        except Exception:
+            pass
+        # Do not leave unread request-body bytes on an HTTP/1.1 keep-alive
+        # socket. Draining an arbitrary oversized body can tie up a worker;
+        # closing the connection after the bounded read preserves framing for
+        # the next request without turning diagnostics into a slow-drain sink.
+        try:
+            handler.close_connection = True
+        except Exception:
+            pass
+        return {"event": "discarded", "reason": "body_too_large"}
+    raw = handler.rfile.read(length) if length else b"{}"
+    try:
+        decoded = raw.decode("utf-8")
+        payload = json.loads(decoded)
+    except Exception:
+        return {"event": "invalid", "reason": "invalid_json"}
+    return payload if isinstance(payload, dict) else {"event": "invalid", "reason": "not_object"}
+
+
+def _handle_client_event_log(handler, body: dict) -> bool:
+    if _client_event_rate_limited(handler):
+        _CLIENT_EVENT_LOGGER.warning(
+            "Dropped client event from %s: rate limit exceeded",
+            _client_ip_for_rate_limit(handler),
+        )
+        return j(handler, {"ok": False, "error": "rate_limited"}, status=429) or True
+    payload = _sanitize_client_event_payload(body)
+    _CLIENT_EVENT_LOGGER.info("Client event from %s: %s", _client_ip_for_rate_limit(handler), payload)
+    return j(handler, {"ok": True, "event": payload.get("event")}) or True
+
+
 def _normalize_provider_id(value: str | None) -> str:
     raw = str(value or "").strip().lower()
     if not raw:
@@ -4708,6 +4836,11 @@ def handle_post(handler, parsed) -> bool:
     if parsed.path == "/api/transcribe":
         return handle_transcribe(handler)
 
+    if parsed.path == "/api/client-events/log":
+        if diag:
+            diag.stage("read_client_event_body")
+        return _handle_client_event_log(handler, _read_client_event_payload(handler))
+
     if diag:
         diag.stage("read_body")
     try:

docs/troubleshooting.md

@@ -93,7 +93,7 @@ If after running steps 1-4 the import still fails *and* `pip install -e .` succe
 
 **Interruption classes.** The WebUI now keeps the user-facing cases separate instead of implying every stale stream was a restart:
 
-- **Browser/SSE connection interrupted** — the live browser `EventSource` transport dropped. The UI reports `Connection interrupted` and tries status/replay/session restore before showing the final browser-side notice.
+- **Browser/SSE connection interrupted** — the live browser `EventSource` transport dropped. The UI reports `Connection interrupted` and tries status/replay/session restore before showing the final browser-side notice. Chat and gateway SSE errors also POST a small sanitized diagnostic event to `/api/client-events/log` (source, session id, stream id, readyState, visibility, online state, path without query string) so server logs can distinguish browser transport loss from backend worker loss.
 - **Lost worker bookkeeping** — the stream id is gone and the worker registry no longer has an active run. Recovery markers carry `interruption_cause: "lost_worker_bookkeeping"` and `/api/chat/stream/status` reports `terminal_state: "lost-worker-bookkeeping"` for non-terminal journals that are no longer active.
 - **Stream/run split-brain** — the stream is gone but `ACTIVE_RUNS` still lists the worker. Recovery markers carry `interruption_cause: "stream_run_split_brain"` so the transcript says this is a bookkeeping split-brain rather than a restart.
 - **Process crash/restart** — `SERVER_START_TIME` is newer than `pending_started_at`, meaning the WebUI process started after the turn began. Recovery markers carry `interruption_cause: "process_restart"` and explicitly say the process-start evidence points to a crash or restart.

static/messages.js

@@ -2080,13 +2080,14 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){
     });
 
     source.addEventListener('error',async e=>{
-      source.close();
-      if(_deferStreamErrorIfOffline()) return;
-      if(_deferStreamErrorIfPageHidden()) return;
       if(_terminalStateReached || _streamFinalized){
         _closeSource();
         return;
       }
+      if(typeof recordClientSSEError==='function') recordClientSSEError('chat-response',{ready_state:source?source.readyState:null,session_id:activeSid,stream_id:streamId,reason:'chat EventSource.onerror'});
+      source.close();
+      if(_deferStreamErrorIfOffline()) return;
+      if(_deferStreamErrorIfPageHidden()) return;
       // Attempt one reconnect if the stream is still active server-side
       if(!_reconnectAttempted && streamId){
         _reconnectAttempted=true;

static/sessions.js

@@ -2438,6 +2438,7 @@ function startGatewaySSE(){
       }catch(e){ /* ignore parse errors */ }
     });
     _gatewaySSE.onerror = () => {
+      if(typeof recordClientSSEError==='function') recordClientSSEError('gateway-sessions',{ready_state:_gatewaySSE?_gatewaySSE.readyState:null,reason:'gateway EventSource.onerror'});
       if(_gatewaySSE){
         _gatewaySSE.close();
         _gatewaySSE = null;

static/workspace.js

@@ -85,6 +85,23 @@ async function api(path,opts={}){
   throw lastErr;
 }
 
+function recordClientSSEError(source, details={}){
+  try{
+    const payload={
+      event:'sse_error',
+      source:String(source||'unknown'),
+      ready_state:details.ready_state,
+      session_id:details.session_id||null,
+      stream_id:details.stream_id||null,
+      visibility_state:(typeof document!=='undefined'&&document.visibilityState)||'unknown',
+      online:(typeof navigator!=='undefined'&&typeof navigator.onLine==='boolean')?navigator.onLine:null,
+      url_path:(typeof location!=='undefined'&&location.pathname)||'/',
+      reason:details.reason||'EventSource.onerror',
+    };
+    void api('/api/client-events/log',{method:'POST',body:JSON.stringify(payload),timeoutMs:3000}).catch(()=>{});
+  }catch(_){}
+}
+
 // Persist/restore expanded directory state per workspace in localStorage
 function _wsExpandKey(){
   const ws=S.session&&S.session.workspace;

tests/test_client_event_diagnostics.py

@@ -0,0 +1,158 @@
+"""Regression coverage for browser-side SSE disconnect diagnostics.
+
+When an EventSource fails in the browser, the server normally only sees a dead
+socket or follow-up probe. Persisting a small, sanitized client event makes the
+next incident diagnosable without logging prompt text or credentials.
+"""
+from pathlib import Path
+from io import BytesIO
+from types import SimpleNamespace
+
+import api.routes as routes
+
+
+REPO = Path(__file__).resolve().parents[1]
+WORKSPACE_JS = (REPO / "static" / "workspace.js").read_text(encoding="utf-8")
+SESSIONS_JS = (REPO / "static" / "sessions.js").read_text(encoding="utf-8")
+MESSAGES_JS = (REPO / "static" / "messages.js").read_text(encoding="utf-8")
+
+
+def test_client_event_log_sanitizes_and_whitelists_fields():
+    payload = {
+        "event": "sse_error",
+        "source": "gateway-sessions",
+        "session_id": "abc123",
+        "stream_id": "stream456",
+        "ready_state": 2,
+        "visibility_state": "hidden",
+        "online": True,
+        "url_path": "/chat/abc123?debug=1",
+        "reason": "onerror",
+        "password": "must-not-survive",
+        "content": "prompt text must not survive",
+        "cookie": "session=secret",
+    }
+
+    sanitized = routes._sanitize_client_event_payload(payload)
+
+    assert sanitized == {
+        "event": "sse_error",
+        "source": "gateway-sessions",
+        "session_id": "abc123",
+        "stream_id": "stream456",
+        "ready_state": 2,
+        "visibility_state": "hidden",
+        "online": True,
+        "url_path": "/chat/abc123",
+        "reason": "onerror",
+    }
+    assert "secret" not in repr(sanitized)
+    assert "prompt text" not in repr(sanitized)
+
+
+def test_client_event_log_bounds_untrusted_values():
+    payload = {
+        "event": "x" * 200,
+        "source": "chat" * 100,
+        "session_id": "s" * 200,
+        "stream_id": "t" * 200,
+        "ready_state": "not-a-number",
+        "visibility_state": "visible" * 40,
+        "online": "yes",
+        "url_path": "https://example.invalid/path?debug=1",
+        "reason": "network" * 80,
+    }
+
+    sanitized = routes._sanitize_client_event_payload(payload)
+
+    assert sanitized["event"] == "x" * 64
+    assert len(sanitized["source"]) == 80
+    assert len(sanitized["session_id"]) == 128
+    assert len(sanitized["stream_id"]) == 128
+    assert "ready_state" not in sanitized
+    assert sanitized["visibility_state"] == ("visible" * 40)[:32]
+    assert sanitized["online"] is True
+    assert sanitized["url_path"] == "/path"
+    assert len(sanitized["reason"]) == 160
+
+
+def test_client_event_log_route_is_wired(monkeypatch):
+    captured = {}
+
+    body_bytes = b'{"event":"sse_error","source":"chat-response"}'
+
+    def fake_handle(handler, body):
+        captured["body"] = body
+        return True
+
+    monkeypatch.setattr(routes, "_check_csrf", lambda handler: True)
+    monkeypatch.setattr(routes, "_handle_client_event_log", fake_handle)
+
+    handler = SimpleNamespace(headers={"Content-Length": str(len(body_bytes))}, rfile=BytesIO(body_bytes))
+    handled = routes.handle_post(handler, SimpleNamespace(path="/api/client-events/log"))
+
+    assert handled is True
+    assert captured["body"] == {"event": "sse_error", "source": "chat-response"}
+
+
+def test_client_event_log_has_endpoint_specific_body_cap():
+    too_large = b"{" + b"x" * (routes._CLIENT_EVENT_MAX_BODY_BYTES + 32) + b"}"
+    handler = SimpleNamespace(
+        headers={"Content-Length": str(len(too_large))},
+        rfile=BytesIO(too_large),
+        close_connection=False,
+    )
+
+    payload = routes._read_client_event_payload(handler)
+
+    assert payload == {"event": "discarded", "reason": "body_too_large"}
+    assert handler.rfile.tell() == routes._CLIENT_EVENT_MAX_BODY_BYTES
+    assert handler.close_connection is True
+
+
+def test_client_event_log_rate_limits_per_client(monkeypatch):
+    routes._CLIENT_EVENT_RATE_LIMIT.clear()
+    monkeypatch.setattr(routes, "j", lambda handler, payload, status=200: payload.update({"status": status}) or True)
+    handler = SimpleNamespace(client_address=("203.0.113.10", 1234))
+
+    for i in range(routes._CLIENT_EVENT_RATE_LIMIT_MAX):
+        assert routes._handle_client_event_log(handler, {"event": f"sse_error_{i}"}) is True
+
+    limited = {}
+    monkeypatch.setattr(routes, "j", lambda handler, payload, status=200: limited.update(payload, status=status) or True)
+    assert routes._handle_client_event_log(handler, {"event": "sse_error_over_limit"}) is True
+
+    assert limited == {"ok": False, "error": "rate_limited", "status": 429}
+    routes._CLIENT_EVENT_RATE_LIMIT.clear()
+
+
+def test_workspace_js_defines_sanitized_client_sse_error_reporter():
+    helper_start = WORKSPACE_JS.index("function recordClientSSEError")
+    helper_block = WORKSPACE_JS[helper_start:helper_start + 900]
+    assert "api/client-events/log" in helper_block
+    assert "document.visibilityState" in helper_block
+    assert "navigator.onLine" in helper_block
+    assert "location.pathname" in helper_block
+    assert "location.search" not in helper_block
+
+
+def test_sessions_js_reports_gateway_sse_errors_with_browser_context():
+    gateway_block_start = SESSIONS_JS.index("_gatewaySSE.onerror = () =>")
+    gateway_block = SESSIONS_JS[gateway_block_start:gateway_block_start + 400]
+    assert "recordClientSSEError('gateway-sessions'" in gateway_block
+    assert "probeGatewaySSEStatus" in gateway_block
+
+
+def test_messages_js_reports_chat_sse_errors_with_stream_identity():
+    error_block_start = MESSAGES_JS.index("source.addEventListener('error',async e=>")
+    error_block = MESSAGES_JS[error_block_start:error_block_start + 900]
+    assert "recordClientSSEError('chat-response'" in error_block
+    assert "session_id:activeSid" in error_block
+    assert "stream_id:streamId" in error_block
+
+
+def test_messages_js_keeps_finalized_stream_guard_before_diagnostic_report():
+    error_block_start = MESSAGES_JS.index("source.addEventListener('error',async e=>")
+    error_block = MESSAGES_JS[error_block_start:error_block_start + 900]
+    assert "_streamFinalized" in error_block
+    assert error_block.index("_streamFinalized") < error_block.index("recordClientSSEError")