From f39aaf56b24db313766074e464270f3cdbc6d04f Mon Sep 17 00:00:00 2001 From: muhammedrazalak Date: Fri, 13 Nov 2026 12:03:47 +0530 Subject: [PATCH] [2093] feat: Add instrumentation support for openai decision model apis --- .../libraries/openai/__init__.py | 34 ++ .../instrumentation/libraries/openai/utils.py | 121 ++++++ .../libraries/openai/wrappers.py | 64 +++ tests/test_openai_instrumentation.py | 363 +++++++++++++++++- 4 files changed, 577 insertions(+), 5 deletions(-) diff --git a/netra/instrumentation/libraries/openai/__init__.py b/netra/instrumentation/libraries/openai/__init__.py index 6288b5d1..3d6986c6 100644 --- a/netra/instrumentation/libraries/openai/__init__.py +++ b/netra/instrumentation/libraries/openai/__init__.py @@ -8,9 +8,11 @@ from netra.instrumentation.libraries.openai.version import __version__ from netra.instrumentation.libraries.openai.wrappers import ( achat_wrapper, + adecisions_wrapper, aembeddings_wrapper, aresponses_wrapper, chat_wrapper, + decisions_wrapper, embeddings_wrapper, responses_wrapper, ) @@ -26,6 +28,10 @@ # ``__name__``, which would silently change the moment this file moves. _TRACER_NAME = "netra.instrumentation.openai" +# The Decisions API only exists in newer openai releases, so its absence is expected +# on older installs and must not be reported as an instrumentation failure. +_DECISIONS_MODULE = "openai.resources.decisions" + class NetraOpenAIInstrumentor(BaseInstrumentor): # type: ignore[misc] """ @@ -87,6 +93,26 @@ def _instrument(self, **kwargs) -> Any: # type: ignore[no-untyped-def] except Exception as e: logger.error(f"Failed to instrument responses: {e}") + self._instrument_decisions(tracer) + + @staticmethod + def _instrument_decisions(tracer: Any) -> None: + """Wrap ``decisions.create`` (sync and async) when the installed openai has it. + + Args: + tracer (Any): The tracer the decision spans are started from. + + Returns: + None + """ + try: + wrap_function_wrapper(_DECISIONS_MODULE, "Decisions.create", decisions_wrapper(tracer)) + wrap_function_wrapper(_DECISIONS_MODULE, "AsyncDecisions.create", adecisions_wrapper(tracer)) + except (AttributeError, ImportError): + logger.debug("openai Decisions API not available in the installed openai version; skipping") + except Exception as e: + logger.error(f"Failed to instrument decisions: {e}") + def _uninstrument(self, **kwargs): # type: ignore[no-untyped-def] """Uninstrument OpenAI client methods""" @@ -113,3 +139,11 @@ def _uninstrument(self, **kwargs): # type: ignore[no-untyped-def] unwrap("openai.resources.responses", "AsyncResponses.create") except (AttributeError, ModuleNotFoundError): logger.error("Failed to uninstrument responses") + + try: + # unwrap() takes the dotted path of the class that owns the method; given + # ``(module, "Class.method")`` it silently does nothing. + unwrap(f"{_DECISIONS_MODULE}.Decisions", "create") + unwrap(f"{_DECISIONS_MODULE}.AsyncDecisions", "create") + except (AttributeError, ImportError): + logger.debug("openai Decisions API not available in the installed openai version; nothing to unwrap") diff --git a/netra/instrumentation/libraries/openai/utils.py b/netra/instrumentation/libraries/openai/utils.py index 493b69f9..2f9ead08 100644 --- a/netra/instrumentation/libraries/openai/utils.py +++ b/netra/instrumentation/libraries/openai/utils.py @@ -30,6 +30,13 @@ # Tool types whose name/description/schema live under a key named after the type in # Chat Completions (``{"type": "function", "function": {...}}``) but flat in the Responses API. _NAMED_TOOL_TYPES = ("function", "custom") +# Decisions API (``client.decisions.create``): the questions asked about the input and +# the typed answers returned for them, each recorded as one JSON array. +GEN_AI_DECISION_QUESTIONS = "gen_ai.decision.questions" +GEN_AI_DECISION_ANSWERS = "gen_ai.decision.answers" +# Decision input item types that carry a tool invocation, mapped to the key holding its arguments. +_DECISION_TOOL_CALL_ARGUMENT_KEYS: Dict[str, str] = {"function_call": "arguments", "custom_tool_call": "input"} +_DECISION_TOOL_OUTPUT_TYPES = ("function_call_output", "custom_tool_call_output") def should_suppress_instrumentation() -> bool: @@ -86,10 +93,101 @@ def set_request_attributes(span: Span, kwargs: Dict[str, Any], operation_type: s _set_chat_completion_input(span, kwargs.get("messages")) elif operation_type == "response": _set_chat_response_input(span, kwargs) + elif operation_type == "decision": + _set_decision_request_attributes(span, kwargs) _set_tool_attributes(span, kwargs) +def _set_decision_request_attributes(span: Span, kwargs: Dict[str, Any]) -> None: + """Record a Decisions API request's input and questions without ever failing the call. + + Runs before the OpenAI call and serialises caller-supplied content, so any + failure is logged and the remaining attributes are left unset instead of + aborting the user's request. + """ + try: + _set_decision_input(span, kwargs.get("input")) + except Exception: # noqa: BLE001 - input capture is best-effort; never block the user's request + logger.warning("netra.instrumentation.libraries.openai: failed to record decision input", exc_info=True) + + try: + _set_decision_questions(span, kwargs.get("questions")) + except Exception: # noqa: BLE001 - question capture is best-effort; never block the user's request + logger.warning("netra.instrumentation.libraries.openai: failed to record decision questions", exc_info=True) + + +def _as_prompt_content(content: Any) -> str: + """Render message content as a string: text as-is, content parts as JSON.""" + if isinstance(content, str): + return content + return json.dumps(content, default=str) + + +def _set_decision_input(span: Span, input_data: Any) -> None: + """Record the Decisions API ``input`` as ``gen_ai.prompt.*`` entries. + + ``input`` is either a string (recorded as one user message) or a list of + messages, function / custom tool calls and their outputs. Tool calls are + recorded as assistant entries holding ``{"name", "arguments"}`` and tool + outputs as ``tool`` entries carrying their ``call_id``, matching how the + Responses API input is recorded. Only lists and tuples are walked: the SDK + accepts any iterable, and consuming a one-shot iterator here would empty + the request the caller is about to send. + """ + if isinstance(input_data, str): + if input_data: + span.set_attribute(f"{SpanAttributes.LLM_PROMPTS}.0.role", "user") + span.set_attribute(f"{SpanAttributes.LLM_PROMPTS}.0.content", input_data) + return + + if not isinstance(input_data, (list, tuple)): + return + + prompt_index = 0 + for item in input_data: + if not isinstance(item, dict): + item = model_as_dict(item) + if not item: + continue + + item_type = item.get("type") or "message" + if item_type in _DECISION_TOOL_CALL_ARGUMENT_KEYS: + role = "assistant" + content = json.dumps( + {"name": item.get("name", ""), "arguments": item.get(_DECISION_TOOL_CALL_ARGUMENT_KEYS[item_type], "")} + ) + elif item_type in _DECISION_TOOL_OUTPUT_TYPES: + if not (output := item.get("output")): + continue + role = "tool" + content = _as_prompt_content(output) + else: + if not (message_content := item.get("content")): + continue + role = item.get("role", "user") + content = _as_prompt_content(message_content) + + span.set_attribute(f"{SpanAttributes.LLM_PROMPTS}.{prompt_index}.role", role) + span.set_attribute(f"{SpanAttributes.LLM_PROMPTS}.{prompt_index}.content", content) + if item_type != "message" and (call_id := item.get("call_id")): + span.set_attribute(f"{SpanAttributes.LLM_PROMPTS}.{prompt_index}.tool_call_id", call_id) + prompt_index += 1 + + +def _set_decision_questions(span: Span, questions: Any) -> None: + """Record the Decisions API ``questions`` as one JSON array under ``gen_ai.decision.questions``. + + Questions are recorded as given (``type``, ``instructions``, ``name`` and the + ``choices`` / ``levels`` they offer). As with ``input``, only lists and tuples + are read so a one-shot iterable reaches the API intact. + """ + if not isinstance(questions, (list, tuple)) or not questions: + return + serialised = [question if isinstance(question, dict) else model_as_dict(question) for question in questions] + span.set_attribute(GEN_AI_DECISION_QUESTIONS, json.dumps(serialised, default=str)) + + def _set_tool_attributes(span: Span, kwargs: Dict[str, Any]) -> None: """Record ``tools`` and ``tool_choice`` without ever failing the request. @@ -259,6 +357,29 @@ def set_response_attributes(span: Span, response_dict: Dict[str, Any]) -> None: _set_custom_usage_attribute(span, usage, response_dict.get("model") or "") _set_response_message_attributes(span, response_dict) + _set_decision_answer_attributes(span, response_dict) + + +def _set_decision_answer_attributes(span: Span, response_dict: Dict[str, Any]) -> None: + """Record a Decisions API response's ``answers``, in question order. + + The answers are recorded twice: as one JSON array under + ``gen_ai.decision.answers`` for querying, and as the span's single assistant + completion so they show up alongside the prompt like any other model output. + Each answer is a predicate, choice, score or refusal object, as returned. + Responses without an ``answers`` list (every other OpenAI API) are ignored. + """ + answers = response_dict.get("answers") + if not isinstance(answers, list): + return + try: + serialised = json.dumps(answers, default=str) + except Exception: # noqa: BLE001 - runs after the call succeeded; never cost the user their response + logger.warning("netra.instrumentation.libraries.openai: failed to record decision answers", exc_info=True) + return + span.set_attribute(GEN_AI_DECISION_ANSWERS, serialised) + span.set_attribute(f"{SpanAttributes.LLM_COMPLETIONS}.0.role", "assistant") + span.set_attribute(f"{SpanAttributes.LLM_COMPLETIONS}.0.content", serialised) def _first_present(usage: Dict[str, Any], *keys: str) -> Any: diff --git a/netra/instrumentation/libraries/openai/wrappers.py b/netra/instrumentation/libraries/openai/wrappers.py index a6e0e07a..de65a50f 100644 --- a/netra/instrumentation/libraries/openai/wrappers.py +++ b/netra/instrumentation/libraries/openai/wrappers.py @@ -22,6 +22,7 @@ CHAT_SPAN_NAME = "openai.chat" EMBEDDING_SPAN_NAME = "openai.embedding" RESPONSE_SPAN_NAME = "openai.response" +DECISION_SPAN_NAME = "openai.decision" TIME_TO_FIRST_TOKEN = "gen_ai.performance.time_to_first_token" RELATIVE_TIME_TO_FIRST_TOKEN = "gen_ai.performance.relative_time_to_first_token" LLM_RESPONSE_DURATION = "llm.response.duration" @@ -278,6 +279,69 @@ async def wrapper(wrapped: Callable[..., Awaitable[Any]], instance: Any, args: A return wrapper +def decisions_wrapper(tracer: Tracer) -> Callable[..., Any]: + """Wrapper for decisions.create (Decisions API). + + The Decisions API has no streaming mode, so the whole call is one span. + """ + + def wrapper(wrapped: Callable[..., Any], instance: Any, args: Tuple[Any, ...], kwargs: Dict[str, Any]) -> Any: + if should_suppress_instrumentation(): + return wrapped(*args, **kwargs) + + with tracer.start_as_current_span( + DECISION_SPAN_NAME, kind=SpanKind.CLIENT, attributes={"llm.request.type": "decision"} + ) as span: + try: + set_request_attributes(span, kwargs, "decision") + response = wrapped(*args, **kwargs) + end_time = time.time() + response_dict = model_as_dict(response) + set_response_attributes(span, response_dict) + record_span_timing(span, LLM_RESPONSE_DURATION, end_time) + record_span_timing(span, TIME_TO_FIRST_TOKEN, end_time, record_event_timestamp=True) + record_span_timing(span, RELATIVE_TIME_TO_FIRST_TOKEN, end_time, use_root_span=True) + span.set_status(Status(StatusCode.OK)) + return response + except Exception as e: + logger.error("netra.instrumentation.libraries.openai: %s", e) + span.set_status(Status(StatusCode.ERROR, str(e))) + raise + + return wrapper + + +def adecisions_wrapper(tracer: Tracer) -> Callable[..., Awaitable[Any]]: + """Async wrapper for decisions.create (Decisions API).""" + + async def wrapper( + wrapped: Callable[..., Awaitable[Any]], instance: Any, args: Tuple[Any, ...], kwargs: Dict[str, Any] + ) -> Any: + if should_suppress_instrumentation(): + return await wrapped(*args, **kwargs) + + with tracer.start_as_current_span( + DECISION_SPAN_NAME, kind=SpanKind.CLIENT, attributes={"llm.request.type": "decision"} + ) as span: + try: + set_request_attributes(span, kwargs, "decision") + response = await wrapped(*args, **kwargs) + end_time = time.time() + response_dict = model_as_dict(response) + set_response_attributes(span, response_dict) + record_span_timing(span, LLM_RESPONSE_DURATION, end_time) + record_span_timing(span, TIME_TO_FIRST_TOKEN, end_time, record_event_timestamp=True) + record_span_timing(span, RELATIVE_TIME_TO_FIRST_TOKEN, end_time, use_root_span=True) + span.set_status(Status(StatusCode.OK)) + return response + except Exception as e: + logger.error("netra.instrumentation.libraries.openai: %s", e) + span.set_status(Status(StatusCode.ERROR, str(e))) + raise + + return wrapper + + class StreamingWrapper(ObjectProxy): # type: ignore[misc] """Wrapper for streaming responses""" diff --git a/tests/test_openai_instrumentation.py b/tests/test_openai_instrumentation.py index c2001edd..8334cb61 100644 --- a/tests/test_openai_instrumentation.py +++ b/tests/test_openai_instrumentation.py @@ -46,8 +46,8 @@ def test_instrument_with_default_parameters(self, mock_wrap_function, mock_get_t instrumentor._instrument() mock_get_tracer.assert_called_once() - # chat x2, embeddings x2, responses x2 - assert mock_wrap_function.call_count == 6 + # chat x2, embeddings x2, responses x2, decisions x2 + assert mock_wrap_function.call_count == 8 @patch("netra.instrumentation.libraries.openai.get_tracer") @patch("netra.instrumentation.libraries.openai.wrap_function_wrapper") @@ -63,7 +63,7 @@ def test_instrument_with_custom_tracer_provider(self, mock_wrap_function, mock_g mock_get_tracer.assert_called_once_with( "netra.instrumentation.openai", mock_get_tracer.call_args[0][1], mock_tracer_provider ) - assert mock_wrap_function.call_count == 6 + assert mock_wrap_function.call_count == 8 @patch("netra.instrumentation.libraries.openai.unwrap") def test_uninstrument(self, mock_unwrap): @@ -72,8 +72,8 @@ def test_uninstrument(self, mock_unwrap): instrumentor._uninstrument() - # chat x2, completions x2, embeddings x2, responses x2 - assert mock_unwrap.call_count == 8 + # chat x2, completions x2, embeddings x2, responses x2, decisions x2 + assert mock_unwrap.call_count == 10 class TestWrappers: @@ -936,3 +936,356 @@ def set_attribute(key, value): assert captured[SpanAttributes.LLM_USAGE_CACHE_CREATION_INPUT_TOKENS] == 7000 assert captured[SpanAttributes.LLM_USAGE_TOTAL_TOKENS] == 12500 assert captured[f"{SpanAttributes.LLM_COMPLETIONS}.0.content"] == "ok" + + +DECISION_USAGE = { + "input_tokens": 1200, + "input_tokens_details": {"cached_tokens": 1000, "cache_write_tokens": 150}, + "output_tokens": 12, + "output_tokens_details": {"reasoning_tokens": 4}, + "total_tokens": 1212, +} + +DECISION_ANSWERS = [ + {"type": "predicate", "name": "is_spam", "probability": 0.08}, + { + "type": "choice", + "name": "category", + "choice": "billing", + "confidence": 0.91, + "probabilities": [{"value": "billing", "probability": 0.91}, {"value": "tech", "probability": 0.09}], + }, + { + "type": "score", + "name": "urgency", + "score": 2.4, + "confidence": 0.7, + "probabilities": [{"label": "low", "value": 1, "probability": 0.2}], + }, + {"type": "refusal", "name": None}, +] + +DECISION_QUESTIONS = [ + {"type": "predicate", "name": "is_spam", "instructions": "Is this message spam?"}, + { + "type": "choice", + "name": "category", + "instructions": "Which team should handle this?", + "choices": [{"value": "billing"}, {"value": "tech", "description": "Technical issues"}], + }, + { + "type": "score", + "name": "urgency", + "instructions": "How urgent is this?", + "levels": [{"label": "low"}, {"label": "medium"}, {"label": "high"}], + }, +] + + +def _recording_span(): + """Return a mock recording span plus the dict its set_attribute calls write into.""" + span = Mock() + span.is_recording.return_value = True + captured: dict = {} + span.set_attribute.side_effect = lambda key, value: captured.__setitem__(key, value) + return span, captured + + +class TestDecisionsInstrumentor: + """Registration of the Decisions API wrappers on NetraOpenAIInstrumentor.""" + + @patch("netra.instrumentation.libraries.openai.get_tracer") + @patch("netra.instrumentation.libraries.openai.wrap_function_wrapper") + def test_decisions_create_is_wrapped_sync_and_async(self, mock_wrap_function, mock_get_tracer): + NetraOpenAIInstrumentor()._instrument() + + targets = [(c.args[0], c.args[1]) for c in mock_wrap_function.call_args_list] + assert ("openai.resources.decisions", "Decisions.create") in targets + assert ("openai.resources.decisions", "AsyncDecisions.create") in targets + + @patch("netra.instrumentation.libraries.openai.logger") + @patch("netra.instrumentation.libraries.openai.get_tracer") + @patch("netra.instrumentation.libraries.openai.wrap_function_wrapper") + def test_openai_without_decisions_is_skipped_without_error(self, mock_wrap_function, mock_get_tracer, mock_logger): + """Older openai releases have no decisions module; that must not be logged as a failure.""" + + def wrap(module, name, wrapper): + if module == "openai.resources.decisions": + raise ModuleNotFoundError("No module named 'openai.resources.decisions'") + + mock_wrap_function.side_effect = wrap + + NetraOpenAIInstrumentor()._instrument() + + mock_logger.error.assert_not_called() + # The other six targets were still wrapped. + assert mock_wrap_function.call_count == 7 + + @patch("netra.instrumentation.libraries.openai.unwrap") + def test_uninstrument_unwraps_decision_classes(self, mock_unwrap): + """unwrap() needs the owning class path; ``(module, "Class.method")`` would be a silent no-op.""" + NetraOpenAIInstrumentor()._uninstrument() + + targets = [(c.args[0], c.args[1]) for c in mock_unwrap.call_args_list] + assert ("openai.resources.decisions.Decisions", "create") in targets + assert ("openai.resources.decisions.AsyncDecisions", "create") in targets + + @patch("netra.instrumentation.libraries.openai.logger") + @patch("netra.instrumentation.libraries.openai.unwrap") + def test_uninstrument_without_decisions_is_quiet(self, mock_unwrap, mock_logger): + def unwrap(obj, attr): + if str(obj).startswith("openai.resources.decisions"): + raise ImportError("No module named 'openai.resources.decisions'") + + mock_unwrap.side_effect = unwrap + + NetraOpenAIInstrumentor()._uninstrument() + + assert not any("ecision" in str(c) for c in mock_logger.error.call_args_list) + + +class TestDecisionRequestAttributes: + """set_request_attributes for operation_type="decision".""" + + @staticmethod + def _capture(kwargs): + from netra.instrumentation.libraries.openai.utils import set_request_attributes + + span, captured = _recording_span() + set_request_attributes(span, kwargs, "decision") + return captured + + def test_model_and_request_type_are_recorded(self): + attrs = self._capture({"model": "gpt-6-luna", "input": "hi", "questions": DECISION_QUESTIONS}) + + assert attrs[SpanAttributes.LLM_REQUEST_TYPE] == "decision" + assert attrs[SpanAttributes.LLM_REQUEST_MODEL] == "gpt-6-luna" + + def test_string_input_is_one_user_prompt(self): + attrs = self._capture({"model": "gpt-6-luna", "input": "Refund me now!", "questions": DECISION_QUESTIONS}) + + assert attrs[f"{SpanAttributes.LLM_PROMPTS}.0.role"] == "user" + assert attrs[f"{SpanAttributes.LLM_PROMPTS}.0.content"] == "Refund me now!" + assert f"{SpanAttributes.LLM_PROMPTS}.1.role" not in attrs + + def test_questions_are_recorded_as_json_array(self): + attrs = self._capture({"model": "gpt-6-luna", "input": "hi", "questions": DECISION_QUESTIONS}) + + assert json.loads(attrs["gen_ai.decision.questions"]) == DECISION_QUESTIONS + + def test_mixed_input_items_are_recorded_in_order(self): + input_items = [ + {"role": "system", "content": "You triage support tickets."}, + { + "role": "user", + "type": "message", + "content": [{"type": "input_text", "text": "Card charged twice"}], + }, + {"type": "function_call", "call_id": "c1", "name": "lookup", "arguments": '{"id": 7}'}, + {"type": "function_call_output", "call_id": "c1", "output": "2 charges found"}, + {"type": "custom_tool_call", "call_id": "c2", "name": "grep", "input": "charge"}, + {"type": "custom_tool_call_output", "call_id": "c2", "output": [{"type": "input_text", "text": "ok"}]}, + {"role": "assistant", "content": "Looking into it."}, + ] + + attrs = self._capture({"model": "gpt-6-luna", "input": input_items, "questions": DECISION_QUESTIONS}) + + p = SpanAttributes.LLM_PROMPTS + assert [attrs[f"{p}.{i}.role"] for i in range(7)] == [ + "system", + "user", + "assistant", + "tool", + "assistant", + "tool", + "assistant", + ] + assert attrs[f"{p}.0.content"] == "You triage support tickets." + assert json.loads(attrs[f"{p}.1.content"]) == [{"type": "input_text", "text": "Card charged twice"}] + assert json.loads(attrs[f"{p}.2.content"]) == {"name": "lookup", "arguments": '{"id": 7}'} + assert attrs[f"{p}.2.tool_call_id"] == "c1" + assert attrs[f"{p}.3.content"] == "2 charges found" + assert attrs[f"{p}.3.tool_call_id"] == "c1" + assert json.loads(attrs[f"{p}.4.content"]) == {"name": "grep", "arguments": "charge"} + assert attrs[f"{p}.5.tool_call_id"] == "c2" + assert f"{p}.0.tool_call_id" not in attrs + assert attrs[f"{p}.6.content"] == "Looking into it." + + def test_one_shot_iterables_are_not_consumed(self): + """The SDK accepts any iterable; reading a generator here would empty the real request.""" + input_items = (m for m in [{"role": "user", "content": "hi"}]) + questions = (q for q in DECISION_QUESTIONS) + + attrs = self._capture({"model": "gpt-6-luna", "input": input_items, "questions": questions}) + + assert len(list(input_items)) == 1 + assert len(list(questions)) == len(DECISION_QUESTIONS) + assert f"{SpanAttributes.LLM_PROMPTS}.0.role" not in attrs + assert "gen_ai.decision.questions" not in attrs + + def test_unserialisable_question_does_not_raise(self): + question: dict = {"type": "predicate", "instructions": "x"} + question["self"] = question + + attrs = self._capture({"model": "gpt-6-luna", "input": "hi", "questions": [question]}) + + assert "gen_ai.decision.questions" not in attrs + assert attrs[f"{SpanAttributes.LLM_PROMPTS}.0.content"] == "hi" + + +class TestDecisionResponseAttributes: + """set_response_attributes for a Decisions API response.""" + + @staticmethod + def _capture(response_dict): + from netra.instrumentation.libraries.openai.utils import set_response_attributes + + span, captured = _recording_span() + set_response_attributes(span, response_dict) + return captured + + def test_answers_are_recorded_as_attribute_and_completion(self): + attrs = self._capture({"model": "gpt-6-luna", "answers": DECISION_ANSWERS, "usage": DECISION_USAGE}) + + assert json.loads(attrs["gen_ai.decision.answers"]) == DECISION_ANSWERS + assert attrs[f"{SpanAttributes.LLM_COMPLETIONS}.0.role"] == "assistant" + assert json.loads(attrs[f"{SpanAttributes.LLM_COMPLETIONS}.0.content"]) == DECISION_ANSWERS + assert attrs[SpanAttributes.LLM_RESPONSE_MODEL] == "gpt-6-luna" + + def test_usage_is_recorded(self): + attrs = self._capture({"model": "gpt-6-luna", "answers": DECISION_ANSWERS, "usage": DECISION_USAGE}) + + assert attrs[SpanAttributes.LLM_USAGE_PROMPT_TOKENS] == 1200 + assert attrs[SpanAttributes.LLM_USAGE_COMPLETION_TOKENS] == 12 + assert attrs[SpanAttributes.LLM_USAGE_CACHE_READ_INPUT_TOKENS] == 1000 + assert attrs[SpanAttributes.LLM_USAGE_CACHE_CREATION_INPUT_TOKENS] == 150 + assert attrs[SpanAttributes.LLM_USAGE_REASONING_TOKENS] == 4 + assert attrs[SpanAttributes.LLM_USAGE_TOTAL_TOKENS] == 1212 + + def test_non_decision_response_records_no_answers(self): + attrs = self._capture({"model": "gpt-4o-mini", "output_text": "ok"}) + + assert "gen_ai.decision.answers" not in attrs + assert attrs[f"{SpanAttributes.LLM_COMPLETIONS}.0.content"] == "ok" + + +class TestDecisionWrappers: + """decisions_wrapper / adecisions_wrapper span lifecycle.""" + + KWARGS = {"model": "gpt-6-luna", "input": "Card charged twice", "questions": DECISION_QUESTIONS} + RESPONSE = {"model": "gpt-6-luna", "answers": DECISION_ANSWERS, "usage": DECISION_USAGE} + + @staticmethod + def _tracer(): + span, captured = _recording_span() + tracer = Mock() + span_context = MagicMock() + span_context.__enter__.return_value = span + tracer.start_as_current_span.return_value = span_context + return tracer, span, captured + + def test_sync_wrapper_records_request_and_response(self): + from netra.instrumentation.libraries.openai.wrappers import DECISION_SPAN_NAME, decisions_wrapper + + tracer, span, attrs = self._tracer() + wrapped = Mock(return_value=self.RESPONSE) + + with patch("netra.instrumentation.libraries.openai.wrappers.record_span_timing"): + result = decisions_wrapper(tracer)(wrapped, Mock(), (), dict(self.KWARGS)) + + assert result is self.RESPONSE + wrapped.assert_called_once_with(**self.KWARGS) + assert tracer.start_as_current_span.call_args.args[0] == DECISION_SPAN_NAME == "openai.decision" + assert tracer.start_as_current_span.call_args.kwargs["attributes"] == {"llm.request.type": "decision"} + assert attrs[f"{SpanAttributes.LLM_PROMPTS}.0.content"] == "Card charged twice" + assert json.loads(attrs["gen_ai.decision.answers"]) == DECISION_ANSWERS + + def test_async_wrapper_records_request_and_response(self): + import asyncio + + from netra.instrumentation.libraries.openai.wrappers import adecisions_wrapper + + tracer, span, attrs = self._tracer() + + async def wrapped(*args, **kwargs): + return self.RESPONSE + + with patch("netra.instrumentation.libraries.openai.wrappers.record_span_timing"): + result = asyncio.run(adecisions_wrapper(tracer)(wrapped, Mock(), (), dict(self.KWARGS))) + + assert result is self.RESPONSE + assert attrs[SpanAttributes.LLM_REQUEST_TYPE] == "decision" + assert json.loads(attrs["gen_ai.decision.answers"]) == DECISION_ANSWERS + + def test_api_error_marks_span_and_reraises(self): + from opentelemetry.trace.status import StatusCode + + from netra.instrumentation.libraries.openai.wrappers import decisions_wrapper + + tracer, span, _ = self._tracer() + wrapped = Mock(side_effect=RuntimeError("400 bad questions")) + + with pytest.raises(RuntimeError, match="400 bad questions"): + decisions_wrapper(tracer)(wrapped, Mock(), (), dict(self.KWARGS)) + + assert span.set_status.call_args.args[0].status_code == StatusCode.ERROR + + @patch("netra.instrumentation.libraries.openai.wrappers.should_suppress_instrumentation", return_value=True) + def test_suppressed_instrumentation_skips_span(self, _): + from netra.instrumentation.libraries.openai.wrappers import decisions_wrapper + + tracer, _, _ = self._tracer() + wrapped = Mock(return_value=self.RESPONSE) + + assert decisions_wrapper(tracer)(wrapped, Mock(), (), dict(self.KWARGS)) is self.RESPONSE + tracer.start_as_current_span.assert_not_called() + + +class TestDecisionsAgainstRealSdk: + """End-to-end through the real openai Decisions resource (skipped on openai releases without it).""" + + def test_sync_and_async_create_emit_decision_spans(self): + import asyncio + + decisions_module = pytest.importorskip("openai.resources.decisions") + from openai.types.decision import Decision + from opentelemetry.sdk.trace import TracerProvider + from opentelemetry.sdk.trace.export import SimpleSpanProcessor + from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter + + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + decision = Decision.model_validate( + {"model": "gpt-6-luna", "answers": DECISION_ANSWERS, "usage": DECISION_USAGE} + ) + + async def apost(*args, **kwargs): + return decision + + # The resources bind ``_post`` to ``client.post`` at construction, so the HTTP layer is the client mock. + sync_client, async_client = Mock(), Mock() + sync_client.post.return_value = decision + async_client.post.side_effect = apost + sync_resource = decisions_module.Decisions(sync_client) + async_resource = decisions_module.AsyncDecisions(async_client) + instrumentor = NetraOpenAIInstrumentor() + instrumentor._instrument(tracer_provider=provider) + try: + assert sync_resource.create(**TestDecisionWrappers.KWARGS) is decision + assert asyncio.run(async_resource.create(**TestDecisionWrappers.KWARGS)) is decision + finally: + instrumentor._uninstrument() + + assert sync_client.post.call_args.args[0] == "/decisions" + + spans = exporter.get_finished_spans() + assert [s.name for s in spans] == ["openai.decision", "openai.decision"] + for span in spans: + assert span.attributes[SpanAttributes.LLM_REQUEST_TYPE] == "decision" + assert span.attributes[SpanAttributes.LLM_RESPONSE_MODEL] == "gpt-6-luna" + assert span.attributes[SpanAttributes.LLM_USAGE_CACHE_CREATION_INPUT_TOKENS] == 150 + answers = json.loads(span.attributes["gen_ai.decision.answers"]) + assert [a["type"] for a in answers] == ["predicate", "choice", "score", "refusal"] + # Uninstrument really removed the wrappers. + assert not hasattr(decisions_module.Decisions.create, "__wrapped__")