feat(responses): add Splunk HEC telemetry to responses endpoint

Wires BackgroundTasks and RH identity through the responses endpoint
handler and adds telemetry hooks for all 6 code paths (streaming and
non-streaming success, error, and shield-blocked events).

- Add background_tasks and rh_identity_context parameters to
  responses_endpoint_handler(), handle_streaming_response(),
  handle_non_streaming_response(), and generate_response()
- Extract RH identity in responses_endpoint_handler() before sub-handler
  dispatch (sub-handlers receive ResponsesRequest, not FastAPI Request)
- Add _queue_responses_splunk_event() helper using BackgroundTasks
- Fire responses_completed, responses_error, and
  responses_shield_blocked sourcetype events on all paths
- Add unit tests covering all 6 telemetry paths

Part 3/3 of RSPEED-2867 Splunk HEC telemetry for /responses.

Signed-off-by: Major Hayden <mhayden@redhat.com>
Signed-off-by: Major Hayden <major@redhat.com>

Author: major

src/app/endpoints/responses.py

@@ -1,4 +1,4 @@
-# pylint: disable=too-many-locals,too-many-branches,too-many-nested-blocks, too-many-arguments,too-many-positional-arguments
+# pylint: disable=too-many-locals,too-many-branches,too-many-nested-blocks,too-many-arguments,too-many-positional-arguments,too-many-lines,too-many-statements
 
 """Handler for REST API call to provide answer using Responses API (LCORE specification)."""
 
@@ -7,7 +7,7 @@
 from datetime import UTC, datetime
 from typing import Annotated, Any, Optional, cast
 
-from fastapi import APIRouter, Depends, HTTPException, Request
+from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request
 from fastapi.responses import StreamingResponse
 from llama_stack_api import (
     OpenAIResponseObject,
@@ -51,6 +51,7 @@
     UnauthorizedResponse,
     UnprocessableEntityResponse,
 )
+from observability import ResponsesEventData, build_responses_event, send_splunk_event
 from utils.conversations import append_turn_items_to_conversation
 from utils.endpoints import (
     check_configuration_loaded,
@@ -87,6 +88,7 @@
     resolve_tool_choice,
     select_model_for_responses,
 )
+from utils.rh_identity import get_rh_identity_context
 from utils.shields import run_shield_moderation
 from utils.suid import (
     normalize_conversation_id,
@@ -130,6 +132,52 @@
 }
 
 
+def _queue_responses_splunk_event(  # pylint: disable=too-many-arguments,too-many-positional-arguments
+    background_tasks: Optional[BackgroundTasks],
+    input_text: str,
+    response_text: str,
+    conversation_id: str,
+    model: str,
+    rh_identity_context: tuple[str, str],
+    inference_time: float,
+    sourcetype: str,
+    input_tokens: int = 0,
+    output_tokens: int = 0,
+) -> None:
+    """Build and queue a Splunk telemetry event for the responses endpoint.
+
+    No-op when background_tasks is None (Splunk telemetry disabled).
+
+    Args:
+        background_tasks: FastAPI background task manager, or None if disabled.
+        input_text: User input text.
+        response_text: Response text from LLM or shield.
+        conversation_id: Conversation identifier.
+        model: Model name used for inference.
+        rh_identity_context: Tuple of (org_id, system_id) from RH identity.
+        inference_time: Request processing duration in seconds.
+        sourcetype: Splunk sourcetype for the event.
+        input_tokens: Number of prompt tokens consumed.
+        output_tokens: Number of completion tokens produced.
+    """
+    if background_tasks is None:
+        return
+    org_id, system_id = rh_identity_context
+    event_data = ResponsesEventData(
+        input_text=input_text,
+        response_text=response_text,
+        conversation_id=conversation_id,
+        model=model,
+        org_id=org_id,
+        system_id=system_id,
+        inference_time=inference_time,
+        input_tokens=input_tokens,
+        output_tokens=output_tokens,
+    )
+    event = build_responses_event(event_data)
+    background_tasks.add_task(send_splunk_event, event, sourcetype)
+
+
 @router.post(
     "/responses",
     responses=responses_response,
@@ -142,6 +190,7 @@ async def responses_endpoint_handler(
     responses_request: ResponsesRequest,
     auth: Annotated[AuthTuple, Depends(get_auth_dependency())],
     mcp_headers: dict[str, dict[str, str]] = Depends(mcp_headers_dependency),
+    background_tasks: BackgroundTasks = cast(BackgroundTasks, None),
 ) -> ResponsesResponse | StreamingResponse:
     """
     Handle request to the /responses endpoint using Responses API (LCORE specification).
@@ -183,6 +232,7 @@ async def responses_endpoint_handler(
         responses_request.instructions, field_name="instructions"
     )
     started_at = datetime.now(UTC)
+    rh_identity_context = get_rh_identity_context(request)
     user_id = auth[0]
 
     await check_mcp_auth(configuration, mcp_headers)
@@ -302,6 +352,8 @@ async def responses_endpoint_handler(
         moderation_result=moderation_result,
         inline_rag_context=inline_rag_context,
         filter_server_tools=filter_server_tools,
+        background_tasks=background_tasks,
+        rh_identity_context=rh_identity_context,
     )
 
 
@@ -314,6 +366,8 @@ async def handle_streaming_response(
     moderation_result: ShieldModerationResult,
     inline_rag_context: RAGContext,
     filter_server_tools: bool = False,
+    background_tasks: BackgroundTasks | None = None,
+    rh_identity_context: tuple[str, str] = ("", ""),
 ) -> StreamingResponse:
     """Handle streaming response from Responses API.
 
@@ -326,6 +380,8 @@ async def handle_streaming_response(
         moderation_result: Result of shield moderation check
         inline_rag_context: Inline RAG context to be used for the response
         filter_server_tools: Whether to filter server-deployed MCP tool events from the stream
+        background_tasks: FastAPI background task manager for telemetry events
+        rh_identity_context: Tuple of (org_id, system_id) from RH identity
     Returns:
         StreamingResponse with SSE-formatted events
     """
@@ -352,6 +408,16 @@ async def handle_streaming_response(
                 user_input=request.input,
                 llm_output=[moderation_result.refusal_response],
             )
+        _queue_responses_splunk_event(
+            background_tasks=background_tasks,
+            input_text=input_text,
+            response_text=moderation_result.message,
+            conversation_id=normalize_conversation_id(api_params.conversation),
+            model=api_params.model,
+            rh_identity_context=rh_identity_context,
+            inference_time=(datetime.now(UTC) - started_at).total_seconds(),
+            sourcetype="responses_shield_blocked",
+        )
     else:
         try:
             response = await client.responses.create(
@@ -368,16 +434,46 @@ async def handle_streaming_response(
             )
         except RuntimeError as e:  # library mode wraps 413 into runtime error
             if is_context_length_error(str(e)):
+                _queue_responses_splunk_event(
+                    background_tasks=background_tasks,
+                    input_text=input_text,
+                    response_text=str(e),
+                    conversation_id=normalize_conversation_id(api_params.conversation),
+                    model=api_params.model,
+                    rh_identity_context=rh_identity_context,
+                    inference_time=(datetime.now(UTC) - started_at).total_seconds(),
+                    sourcetype="responses_error",
+                )
                 error_response = PromptTooLongResponse(model=api_params.model)
                 raise HTTPException(**error_response.model_dump()) from e
             raise e
         except APIConnectionError as e:
+            _queue_responses_splunk_event(
+                background_tasks=background_tasks,
+                input_text=input_text,
+                response_text=str(e),
+                conversation_id=normalize_conversation_id(api_params.conversation),
+                model=api_params.model,
+                rh_identity_context=rh_identity_context,
+                inference_time=(datetime.now(UTC) - started_at).total_seconds(),
+                sourcetype="responses_error",
+            )
             error_response = ServiceUnavailableResponse(
                 backend_name="Llama Stack",
                 cause=str(e),
             )
             raise HTTPException(**error_response.model_dump()) from e
         except (LLSApiStatusError, OpenAIAPIStatusError) as e:
+            _queue_responses_splunk_event(
+                background_tasks=background_tasks,
+                input_text=input_text,
+                response_text=str(e),
+                conversation_id=normalize_conversation_id(api_params.conversation),
+                model=api_params.model,
+                rh_identity_context=rh_identity_context,
+                inference_time=(datetime.now(UTC) - started_at).total_seconds(),
+                sourcetype="responses_error",
+            )
             error_response = handle_known_apistatus_errors(e, api_params.model)
             raise HTTPException(**error_response.model_dump()) from e
 
@@ -391,6 +487,8 @@ async def handle_streaming_response(
             started_at=started_at,
             api_params=api_params,
             generate_topic_summary=request.generate_topic_summary or False,
+            background_tasks=background_tasks,
+            rh_identity_context=rh_identity_context,
         ),
         media_type="text/event-stream",
     )
@@ -742,6 +840,8 @@ async def generate_response(
     started_at: datetime,
     api_params: ResponsesApiParams,
     generate_topic_summary: bool,
+    background_tasks: BackgroundTasks | None = None,
+    rh_identity_context: tuple[str, str] = ("", ""),
 ) -> AsyncIterator[str]:
     """Stream the response from the generator and persist conversation details.
 
@@ -756,6 +856,8 @@ async def generate_response(
         started_at: Timestamp when the conversation started
         api_params: ResponsesApiParams
         generate_topic_summary: Whether to generate topic summary for new conversations
+        background_tasks: FastAPI background task manager for telemetry events
+        rh_identity_context: Tuple of (org_id, system_id) from RH identity
     Yields:
         SSE-formatted strings from the generator
     """
@@ -783,6 +885,22 @@ async def generate_response(
             skip_userid_check=skip_userid_check,
             topic_summary=topic_summary,
         )
+    _queue_responses_splunk_event(
+        background_tasks=background_tasks,
+        input_text=input_text,
+        response_text=turn_summary.llm_response,
+        conversation_id=normalize_conversation_id(api_params.conversation),
+        model=api_params.model,
+        rh_identity_context=rh_identity_context,
+        inference_time=(completed_at - started_at).total_seconds(),
+        sourcetype="responses_completed",
+        input_tokens=(
+            turn_summary.token_usage.input_tokens if turn_summary.token_usage else 0
+        ),
+        output_tokens=(
+            turn_summary.token_usage.output_tokens if turn_summary.token_usage else 0
+        ),
+    )
 
 
 async def handle_non_streaming_response(
@@ -794,6 +912,8 @@ async def handle_non_streaming_response(
     moderation_result: ShieldModerationResult,
     inline_rag_context: RAGContext,
     filter_server_tools: bool = False,
+    background_tasks: BackgroundTasks | None = None,
+    rh_identity_context: tuple[str, str] = ("", ""),
 ) -> ResponsesResponse:
     """Handle non-streaming response from Responses API.
 
@@ -806,6 +926,8 @@ async def handle_non_streaming_response(
         moderation_result: Result of shield moderation check
         inline_rag_context: Inline RAG context to be used for the response
         filter_server_tools: Whether to filter server-deployed MCP tool output
+        background_tasks: FastAPI background task manager for telemetry events
+        rh_identity_context: Tuple of (org_id, system_id) from RH identity
     Returns:
         ResponsesResponse with the completed response
     """
@@ -830,6 +952,16 @@ async def handle_non_streaming_response(
                 user_input=request.input,
                 llm_output=[moderation_result.refusal_response],
             )
+        _queue_responses_splunk_event(
+            background_tasks=background_tasks,
+            input_text=input_text,
+            response_text=output_text,
+            conversation_id=normalize_conversation_id(api_params.conversation),
+            model=api_params.model,
+            rh_identity_context=rh_identity_context,
+            inference_time=(datetime.now(UTC) - started_at).total_seconds(),
+            sourcetype="responses_shield_blocked",
+        )
     else:
         try:
             api_response = cast(
@@ -854,16 +986,46 @@ async def handle_non_streaming_response(
 
         except RuntimeError as e:
             if is_context_length_error(str(e)):
+                _queue_responses_splunk_event(
+                    background_tasks=background_tasks,
+                    input_text=input_text,
+                    response_text=str(e),
+                    conversation_id=normalize_conversation_id(api_params.conversation),
+                    model=api_params.model,
+                    rh_identity_context=rh_identity_context,
+                    inference_time=(datetime.now(UTC) - started_at).total_seconds(),
+                    sourcetype="responses_error",
+                )
                 error_response = PromptTooLongResponse(model=api_params.model)
                 raise HTTPException(**error_response.model_dump()) from e
             raise e
         except APIConnectionError as e:
+            _queue_responses_splunk_event(
+                background_tasks=background_tasks,
+                input_text=input_text,
+                response_text=str(e),
+                conversation_id=normalize_conversation_id(api_params.conversation),
+                model=api_params.model,
+                rh_identity_context=rh_identity_context,
+                inference_time=(datetime.now(UTC) - started_at).total_seconds(),
+                sourcetype="responses_error",
+            )
             error_response = ServiceUnavailableResponse(
                 backend_name="Llama Stack",
                 cause=str(e),
             )
             raise HTTPException(**error_response.model_dump()) from e
         except (LLSApiStatusError, OpenAIAPIStatusError) as e:
+            _queue_responses_splunk_event(
+                background_tasks=background_tasks,
+                input_text=input_text,
+                response_text=str(e),
+                conversation_id=normalize_conversation_id(api_params.conversation),
+                model=api_params.model,
+                rh_identity_context=rh_identity_context,
+                inference_time=(datetime.now(UTC) - started_at).total_seconds(),
+                sourcetype="responses_error",
+            )
             error_response = handle_known_apistatus_errors(e, api_params.model)
             raise HTTPException(**error_response.model_dump()) from e
 
@@ -891,6 +1053,25 @@ async def handle_non_streaming_response(
     )
     turn_summary.rag_chunks.extend(inline_rag_context.rag_chunks)
     completed_at = datetime.now(UTC)
+    if moderation_result.decision != "blocked":
+        _queue_responses_splunk_event(
+            background_tasks=background_tasks,
+            input_text=input_text,
+            response_text=output_text,
+            conversation_id=normalize_conversation_id(api_params.conversation),
+            model=api_params.model,
+            rh_identity_context=rh_identity_context,
+            inference_time=(completed_at - started_at).total_seconds(),
+            sourcetype="responses_completed",
+            input_tokens=(
+                turn_summary.token_usage.input_tokens if turn_summary.token_usage else 0
+            ),
+            output_tokens=(
+                turn_summary.token_usage.output_tokens
+                if turn_summary.token_usage
+                else 0
+            ),
+        )
     if api_params.store:
         store_query_results(
             user_id=user_id,