Add structured timing log to stream_query for slow-request diagnosis

Every SSE request emits a JSON summary line to the logger on completion,
cancellation, or error:

  {"event":"stream_request","conv":"...","user":"ryan",
   "total_ms":45210,"thinking_ms":38100,"ttft_ms":38200,
   "tokens":112,"tools":[{"tool":"web_search","ms":820}],
   "model":"llama","cancelled":false}

Fields:
- total_ms    end-to-end wall time
- thinking_ms time from first LLM call to first content token (reasoning)
- ttft_ms     time to first content token from request start
- tokens      approximate content token count (stream deltas)
- tools       each tool invoked with wall-clock duration
- model       LLM name from LangChain metadata (catches fallback to OpenAI)
- cancelled   true when client disconnected mid-stream

Tool timers use the LangChain run_id to correctly pair start/end even
when the same tool is called multiple times in one turn.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

blueprints/conversation/__init__.py

@@ -219,12 +219,19 @@ async def stream_query():
     async def event_generator():
         final_message = None
         streamed = ""
-        # Reasoning models emit empty-content deltas while thinking; surface a
-        # throttled heartbeat so the UI shows progress (and bytes keep flowing
-        # on the SSE connection) instead of dead air for the whole think.
         last_thinking = 0.0
         turn_has_content = False
 
+        # Timing / observability
+        t0 = time.monotonic()
+        llm_start_at: float | None = None
+        first_content_at: float | None = None
+        tool_timers: dict[str, float] = {}  # run_id -> start time
+        tool_spans: list[dict] = []
+        n_tokens = 0
+        model_name: str | None = None
+        cancelled = False
+
         async def persist(text: str) -> None:
             await add_message_to_conversation(
                 conversation=conversation,
@@ -233,6 +240,35 @@ async def stream_query():
                 user=user,
             )
 
+        def _log_summary(*, error: str | None = None) -> None:
+            now = time.monotonic()
+            total_ms = round((now - t0) * 1000)
+            thinking_ms = (
+                round((first_content_at - llm_start_at) * 1000)
+                if llm_start_at and first_content_at
+                else None
+            )
+            ttft_ms = (
+                round((first_content_at - t0) * 1000) if first_content_at else None
+            )
+            logging.info(
+                json.dumps(
+                    {
+                        "event": "stream_request",
+                        "conv": str(conversation.id),
+                        "user": user.username,
+                        "total_ms": total_ms,
+                        "thinking_ms": thinking_ms,
+                        "ttft_ms": ttft_ms,
+                        "tokens": n_tokens,
+                        "tools": tool_spans,
+                        "model": model_name,
+                        "cancelled": cancelled,
+                        **({"error": error} if error else {}),
+                    }
+                )
+            )
+
         try:
             async for event in main_agent.astream_events(
                 payload, version="v2", config=agent_config
@@ -240,9 +276,19 @@ async def stream_query():
                 event_type = event.get("event")
 
                 if event_type == "on_tool_start":
+                    tool_timers[event["run_id"]] = time.monotonic()
                     yield f"data: {json.dumps({'type': 'tool_start', 'tool': event['name']})}\n\n"
 
                 elif event_type == "on_tool_end":
+                    start = tool_timers.pop(event["run_id"], None)
+                    tool_spans.append(
+                        {
+                            "tool": event["name"],
+                            "ms": round((time.monotonic() - start) * 1000)
+                            if start
+                            else None,
+                        }
+                    )
                     yield f"data: {json.dumps({'type': 'tool_end', 'tool': event['name']})}\n\n"
                     # Persist tool output so later turns keep the fetched context
                     output = event.get("data", {}).get("output")
@@ -256,12 +302,21 @@ async def stream_query():
 
                 elif event_type == "on_chat_model_start":
                     turn_has_content = False
+                    if llm_start_at is None:
+                        llm_start_at = time.monotonic()
+                    if model_name is None:
+                        model_name = event.get("metadata", {}).get(
+                            "ls_model_name"
+                        ) or event.get("name")
 
                 elif event_type == "on_chat_model_stream":
                     chunk = event.get("data", {}).get("chunk")
                     content = getattr(chunk, "content", None)
                     if isinstance(content, str) and content:
+                        if first_content_at is None:
+                            first_content_at = time.monotonic()
                         turn_has_content = True
+                        n_tokens += 1
                         streamed += content
                         yield f"data: {json.dumps({'type': 'content', 'delta': content})}\n\n"
                     elif not turn_has_content:
@@ -281,27 +336,25 @@ async def stream_query():
                                 final_message = content
 
         except asyncio.CancelledError:
-            # The client disconnected mid-stream (e.g. a route change aborts the
+            cancelled = True
-            # fetch). CancelledError is not an Exception, so without this branch
+            _log_summary()
-            # persistence below would be skipped and the answer lost. Save what
-            # was generated in a detached task, then honor the cancellation.
             text = final_message or streamed
             if text:
                 _persist_in_background(persist(text))
             raise
         except Exception as e:
+            _log_summary(error=str(e))
             yield f"data: {json.dumps({'type': 'error', 'message': str(e)})}\n\n"
 
         final_message = final_message or streamed
         if final_message:
-            # Persist via a detached, shielded task so a disconnect at this
-            # point (after generation completes) still commits the answer.
             task = _persist_in_background(persist(final_message))
             await asyncio.shield(task)
             yield f"data: {json.dumps({'type': 'response', 'message': final_message})}\n\n"
         else:
             yield f"data: {json.dumps({'type': 'error', 'message': 'No response generated'})}\n\n"
 
+        _log_summary()
         yield "data: [DONE]\n\n"
 
     return await make_response(