Stream the buddy agent turn live from the AI service

src/api/routes/buddy.py

@@ -21,7 +21,14 @@
 from ingestion.source_state_store import SourceStateStore
 from llm.base import LLMClient, Message, ToolCall, ToolSpec
 from llm.errors import LLMUnavailableError
-from onboarding.buddy_agent import run_agent_turn
+from onboarding.buddy_agent import (
+    AgentReasoning,
+    AgentToken,
+    AgentToolUse,
+    AgentTurnResult,
+    run_agent_turn,
+    stream_agent_turn,
+)
 from onboarding.buddy_compact import compact_memory
 from onboarding.buddy_open import stream_session
 from onboarding.vocabulary import Vocabulary
@@ -42,10 +49,17 @@ def _to_message(schema: BuddyAgentMessageSchema) -> Message:
         ]
     if schema.tool_call_id is not None:
         msg["tool_call_id"] = schema.tool_call_id
+    if schema.reasoning:
+        msg["reasoning"] = schema.reasoning
+    if schema.reasoning_details:
+        msg["reasoning_details"] = [dict(d) for d in schema.reasoning_details]
     return msg
 
 
 def _from_message(msg: Message) -> BuddyAgentMessageSchema:
+    # The reasoning rides along so the hop after a backend tool can hand it back:
+    # a provider with extended thinking rejects a tool turn that lost its signed
+    # thinking blocks.
     return BuddyAgentMessageSchema(
         role=msg["role"],
         content=msg.get("content") or "",
@@ -56,6 +70,8 @@ def _from_message(msg: Message) -> BuddyAgentMessageSchema:
             for call in msg.get("tool_calls") or []
         ],
         tool_call_id=msg.get("tool_call_id"),
+        reasoning=msg.get("reasoning"),
+        reasoning_details=list(msg.get("reasoning_details") or []),
     )
 
 
@@ -67,6 +83,14 @@ def _to_toolspec(schema: BuddyToolSpecSchema) -> ToolSpec:
     )
 
 
+def _vocabulary(body: BuddyAgentRequest) -> Vocabulary:
+    return Vocabulary(
+        contribution_noun=body.vocabulary.contribution_noun,
+        contribution_noun_plural=body.vocabulary.contribution_noun_plural,
+        contribution_verb_past=body.vocabulary.contribution_verb_past,
+    )
+
+
 def _narrowing(body: BuddyAgentRequest) -> RetrievalFilters | None:
     """The reader's source-system and time narrowing, if they chose any.
 
@@ -82,6 +106,29 @@ def _narrowing(body: BuddyAgentRequest) -> RetrievalFilters | None:
     )
 
 
+def _agent_response(result: AgentTurnResult) -> BuddyAgentResponse:
+    return BuddyAgentResponse(
+        final=result.final,
+        text=result.text,
+        messages=[_from_message(m) for m in result.messages],
+        pending_tool_calls=[
+            BuddyToolCallSchema(
+                id=call.id, name=call.name, arguments=dict(call.arguments)
+            )
+            for call in result.pending_tool_calls
+        ],
+        citations=[
+            BuddyCitationSchema(
+                artifact_id=cit.artifact_id,
+                start_line=cit.start_line,
+                start_page=cit.start_page,
+            )
+            for cit in result.citations
+        ],
+        reasoning=result.reasoning,
+    )
+
+
 @router.post(
     "/onboarding/buddy/agent",
     response_model=BuddyAgentResponse,
@@ -104,22 +151,18 @@ def buddy_agent(
     ``capabilities_enabled`` and ``team_mode`` pick the persona. Both are read on every
     hop rather than only the first: the persona is rebuilt each time, so a resume hop
     that omitted one would finish the turn in the other mode.
+
+    The same turn as ``/onboarding/buddy/agent/stream``, buffered into one response.
     """
-    messages = [_to_message(m) for m in body.messages]
-    backend_tools = [_to_toolspec(t) for t in body.backend_tools]
     try:
         result = run_agent_turn(
-            messages,
-            backend_tools,
+            [_to_message(m) for m in body.messages],
+            [_to_toolspec(t) for t in body.backend_tools],
             llm,
             store,
             exclusions=source_state.get_exclusions(),
             prior_summary=body.prior_summary,
-            vocabulary=Vocabulary(
-                contribution_noun=body.vocabulary.contribution_noun,
-                contribution_noun_plural=body.vocabulary.contribution_noun_plural,
-                contribution_verb_past=body.vocabulary.contribution_verb_past,
-            ),
+            vocabulary=_vocabulary(body),
             project_ids=frozenset(body.project_ids),
             capabilities_enabled=body.capabilities_enabled,
             team_mode=body.team_mode,
@@ -130,26 +173,86 @@ def buddy_agent(
             status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=str(exc)
         ) from exc
 
-    return BuddyAgentResponse(
-        final=result.final,
-        text=result.text,
-        messages=[_from_message(m) for m in result.messages],
-        pending_tool_calls=[
-            BuddyToolCallSchema(
-                id=call.id, name=call.name, arguments=dict(call.arguments)
-            )
-            for call in result.pending_tool_calls
-        ],
-        citations=[
-            BuddyCitationSchema(
-                artifact_id=cit.artifact_id,
-                start_line=cit.start_line,
-                start_page=cit.start_page,
+    return _agent_response(result)
+
+
+@router.post(
+    "/onboarding/buddy/agent/stream",
+    summary="Run one agentic buddy turn, streaming it as it happens",
+    response_class=StreamingResponse,
+    tags=["onboarding-buddy"],
+    responses={422: {"model": ValidationErrorResponse}},
+)
+def buddy_agent_stream(
+    body: BuddyAgentRequest,
+    llm: LLMClient = Depends(get_llm),
+    store: VectorStore = Depends(get_store),
+    source_state: SourceStateStore = Depends(get_source_state_store),
+) -> StreamingResponse:
+    """One turn of the tool-using buddy, with the model's work visible as it happens.
+
+    Takes the same request as ``/onboarding/buddy/agent`` and runs the same turn;
+    what differs is that nothing waits for the last hop. Emits ``reasoning`` and
+    ``token`` events, each carrying a ``content`` fragment, as the model produces
+    them; a ``tool_use`` event (``name``, ``arguments``) before each search this
+    service runs itself; and one terminal ``result`` carrying the fields of
+    ``BuddyAgentResponse`` (``final``, ``text``, ``messages``, ``pending_tool_calls``,
+    ``citations``, ``reasoning``). The ``text`` of ``result`` is authoritative: it is
+    what the caller stores and carries back, and the ``token`` events spell it out
+    except where a turn that streamed text for one hop then searched again, which
+    joins its hops with a blank line.
+
+    A model that is unavailable, or any failure mid-turn, becomes a terminal ``error``
+    event rather than a truncated body; the status line is already sent by then.
+    """
+    messages = [_to_message(m) for m in body.messages]
+    backend_tools = [_to_toolspec(t) for t in body.backend_tools]
+    exclusions = source_state.get_exclusions()
+    narrowing = _narrowing(body)
+
+    def event_stream() -> Iterator[str]:
+        try:
+            for event in stream_agent_turn(
+                messages,
+                backend_tools,
+                llm,
+                store,
+                exclusions=exclusions,
+                prior_summary=body.prior_summary,
+                vocabulary=_vocabulary(body),
+                project_ids=frozenset(body.project_ids),
+                capabilities_enabled=body.capabilities_enabled,
+                team_mode=body.team_mode,
+                filters=narrowing,
+            ):
+                if isinstance(event, AgentReasoning):
+                    yield sse_event({"type": "reasoning", "content": event.text})
+                elif isinstance(event, AgentToken):
+                    yield sse_event({"type": "token", "content": event.text})
+                elif isinstance(event, AgentToolUse):
+                    yield sse_event(
+                        {
+                            "type": "tool_use",
+                            "name": event.name,
+                            "arguments": event.arguments,
+                        }
+                    )
+                else:
+                    yield sse_event(
+                        {
+                            "type": "result",
+                            **_agent_response(event).model_dump(mode="json"),
+                        }
+                    )
+        except LLMUnavailableError as exc:
+            yield sse_event({"type": "error", "message": str(exc)})
+        except Exception:
+            logger.exception("Unexpected error in buddy agent stream")
+            yield sse_event(
+                {"type": "error", "message": "An unexpected error occurred"}
             )
-            for cit in result.citations
-        ],
-        reasoning=result.reasoning,
-    )
+
+    return StreamingResponse(event_stream(), media_type="text/event-stream")
 
 
 @router.post(

src/api/schemas.py

@@ -1303,6 +1303,22 @@ class BuddyAgentMessageSchema(BaseModel):
         default=None,
         description="On a tool-result turn, the id of the call it answers.",
     )
+    reasoning: str | None = Field(
+        default=None,
+        description=(
+            "On an assistant turn, the model's plain-text reasoning. Opaque to the "
+            "caller; carry it back verbatim."
+        ),
+    )
+    reasoning_details: list[dict[str, object]] = Field(
+        default_factory=list[dict[str, object]],
+        description=(
+            "On an assistant turn, the provider's structured (possibly signed) "
+            "reasoning blocks. Opaque to the caller; carry them back verbatim, "
+            "because a provider with extended thinking rejects a resumed tool turn "
+            "that lost them."
+        ),
+    )
 
 
 class BuddyToolSpecSchema(BaseModel):