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

Add telemetry events for responses endpoint covering completed,
shield-blocked, and error paths. Error paths use asyncio.create_task()
instead of BackgroundTasks since FastAPI discards background tasks on
non-2xx responses.

RSPEED-2867

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

Author: major

src/app/endpoints/responses.py

@@ -1,13 +1,14 @@
-# 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)."""
 
+import asyncio
 import json
 from collections.abc import AsyncIterator
 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,
@@ -53,6 +54,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,
@@ -89,6 +91,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,
@@ -132,6 +135,60 @@
 }
 
 
+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,
+    fire_and_forget: bool = False,
+) -> None:
+    """Build and queue a Splunk telemetry event for the responses endpoint.
+
+    No-op when background_tasks is None and fire_and_forget is False
+    (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.
+        fire_and_forget: When True, dispatch via asyncio.create_task() instead
+            of background_tasks.  Use for error paths where an HTTPException
+            follows, since FastAPI discards BackgroundTasks on non-2xx responses.
+    """
+    if not fire_and_forget and 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)
+    if fire_and_forget:
+        asyncio.create_task(send_splunk_event(event, sourcetype))
+    elif background_tasks is not None:
+        background_tasks.add_task(send_splunk_event, event, sourcetype)
+
+
 @router.post(
     "/responses",
     responses=responses_response,
@@ -144,6 +201,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).
@@ -187,6 +245,7 @@ async def responses_endpoint_handler(
     )
     instructions_substituted = client_instructions is None
     started_at = datetime.now(UTC)
+    rh_identity_context = get_rh_identity_context(request)
     user_id = auth[0]
 
     await check_mcp_auth(configuration, mcp_headers)
@@ -307,6 +366,8 @@ async def responses_endpoint_handler(
         inline_rag_context=inline_rag_context,
         filter_server_tools=filter_server_tools,
         instructions_substituted=instructions_substituted,
+        background_tasks=background_tasks,
+        rh_identity_context=rh_identity_context,
     )
 
 
@@ -320,6 +381,8 @@ async def handle_streaming_response(
     inline_rag_context: RAGContext,
     filter_server_tools: bool = False,
     instructions_substituted: bool = False,
+    background_tasks: BackgroundTasks | None = None,
+    rh_identity_context: tuple[str, str] = ("", ""),
 ) -> StreamingResponse:
     """Handle streaming response from Responses API.
 
@@ -333,6 +396,8 @@ async def handle_streaming_response(
         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
         instructions_substituted: Whether the server substituted the instructions
+        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
     """
@@ -359,6 +424,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(
@@ -376,16 +451,49 @@ 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",
+                    fire_and_forget=True,
+                )
                 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",
+                fire_and_forget=True,
+            )
             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",
+                fire_and_forget=True,
+            )
             error_response = handle_known_apistatus_errors(e, api_params.model)
             raise HTTPException(**error_response.model_dump()) from e
 
@@ -399,6 +507,9 @@ 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,
+            shield_blocked=(moderation_result.decision == "blocked"),
         ),
         media_type="text/event-stream",
     )
@@ -794,6 +905,9 @@ 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] = ("", ""),
+    shield_blocked: bool = False,
 ) -> AsyncIterator[str]:
     """Stream the response from the generator and persist conversation details.
 
@@ -808,6 +922,9 @@ 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
+        shield_blocked: Whether the request was blocked by a shield
     Yields:
         SSE-formatted strings from the generator
     """
@@ -835,6 +952,25 @@ async def generate_response(
             skip_userid_check=skip_userid_check,
             topic_summary=topic_summary,
         )
+    if not shield_blocked:
+        _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(
@@ -847,6 +983,8 @@ async def handle_non_streaming_response(
     inline_rag_context: RAGContext,
     filter_server_tools: bool = False,
     instructions_substituted: bool = False,
+    background_tasks: BackgroundTasks | None = None,
+    rh_identity_context: tuple[str, str] = ("", ""),
 ) -> ResponsesResponse:
     """Handle non-streaming response from Responses API.
 
@@ -860,6 +998,8 @@ async def handle_non_streaming_response(
         inline_rag_context: Inline RAG context to be used for the response
         filter_server_tools: Whether to filter server-deployed MCP tool output
         instructions_substituted: Whether the server substituted the instructions
+        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
     """
@@ -884,6 +1024,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(
@@ -908,16 +1058,49 @@ 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",
+                    fire_and_forget=True,
+                )
                 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",
+                fire_and_forget=True,
+            )
             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",
+                fire_and_forget=True,
+            )
             error_response = handle_known_apistatus_errors(e, api_params.model)
             raise HTTPException(**error_response.model_dump()) from e
 
@@ -945,6 +1128,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,

src/app/endpoints/rlsapi_v1.py

@@ -78,10 +78,6 @@ class TemplateRenderError(Exception):
 )
 
 
-# Backward-compatible alias so existing test imports continue to work.
-_get_rh_identity_context = get_rh_identity_context
-
-
 infer_responses: dict[int | str, dict[str, Any]] = {
     200: RlsapiV1InferResponse.openapi_response(),
     401: UnauthorizedResponse.openapi_response(examples=UNAUTHORIZED_OPENAPI_EXAMPLES),