Fix/HYBIM-962 review follow ups

CHANGELOG.md

@@ -12,6 +12,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Agent Control spans exported over OTLP now include the control discriminator
   and complete `agent_control.*` field set required for backend classification
   and Controls-card rendering.
+- Explicitly empty message parts are preserved in OTLP telemetry instead of
+  being serialized as the text `"[]"`.
+- OTLP partial-success acknowledgements recognize positive integral JSON-number
+  rejection counts such as `3.0`.
 
 ## [0.1.1] - 2026-08-03
 

src/splunk_ao/converter/attribute_mapping.py

@@ -172,7 +172,10 @@ def _mapped_message(source: dict[str, Any], default_role: str) -> dict[str, Any]
     tool_call_id = source.pop("tool_call_id", None)
     tool_calls = source.pop("tool_calls", None)
 
-    if source_parts is not None:
+    # An explicitly supplied parts field is authoritative over legacy content.
+    if isinstance(source_parts, list) and not source_parts:
+        parts = []
+    elif source_parts is not None:
         parts = _content_parts(source_parts)
     elif role == "tool":
         response = {"type": "tool_call_response", "response": _parse_json_value(content)}

src/splunk_ao/decorator.py

@@ -757,9 +757,7 @@ def _prepare_call(
         set_trace_context = False
         if not existing_trace:
             if current_parent is not None:
-                trace = current_parent
-                while trace._parent is not None:
-                    trace = trace._parent
+                trace = client_instance._current_root()
                 if not isinstance(trace, Trace):
                     raise RuntimeError("Active Splunk AO operation does not have a trace root")
             else:
@@ -903,12 +901,7 @@ def _complete_call(
         return result
 
     def _conclude_owned_trace(self, call_state: _CallState, output: Any, status_code: int | None) -> None:
-        current_parent = call_state.logger.current_parent()
-        root = current_parent
-        while root is not None and root._parent is not None:
-            root = root._parent
-
-        if root is not call_state.trace:
+        if not call_state.logger._is_current_root(call_state.trace):
             return
 
         try:

src/splunk_ao/exporter/diagnostics.py

@@ -4,6 +4,7 @@
 
 import json
 import logging
+import math
 import re
 import threading
 import time
@@ -28,6 +29,7 @@
 _MAX_RESPONSE_BYTES = 64 * 1024
 _MAX_MESSAGE_LENGTH = 512
 _MAX_REJECTION_KEYS = 8
+_MAX_SAFE_JSON_INTEGER = 2**53 - 1
 _DEFAULT_LOG_INTERVAL_SECONDS = 60.0
 _SAFE_KEY = re.compile(r"[^a-zA-Z0-9_.-]+")
 _PROTOBUF_CONTENT_TYPES = frozenset(
@@ -247,6 +249,14 @@ def _positive_json_integer(value: object) -> int | None:
         return None
     if isinstance(value, int):
         return value if value > 0 else None
+    if (
+        isinstance(value, float)
+        and math.isfinite(value)
+        and value.is_integer()
+        and abs(value) <= _MAX_SAFE_JSON_INTEGER
+    ):
+        parsed = int(value)
+        return parsed if parsed > 0 else None
     if isinstance(value, str) and value.isdecimal():
         parsed = int(value)
         return parsed if parsed > 0 else None

src/splunk_ao/handlers/agent_control/bridge.py

@@ -245,11 +245,8 @@ def _active_context(self) -> _ActiveContext | None:
         if current_parent is None or current_parent.id is None:
             return None
 
-        root_parent = current_parent
-        while getattr(root_parent, "_parent", None) is not None:
-            root_parent = root_parent._parent
-
-        if getattr(root_parent, "id", None) is None:
+        root_parent = self._splunk_ao_logger._current_root()
+        if root_parent is None or root_parent.id is None:
             return None
 
         return _ActiveContext(trace_id=str(root_parent.id), span_id=str(current_parent.id))

src/splunk_ao/handlers/base_handler.py

@@ -105,12 +105,7 @@ def commit(self) -> None:
             self._root_node = None
 
     def _conclude_owned_trace(self, trace: Any, output: Any, status_code: int | None) -> None:
-        current_parent = self._splunk_ao_logger.current_parent()
-        root = current_parent
-        while root is not None and root._parent is not None:
-            root = root._parent
-
-        if root is trace:
+        if self._splunk_ao_logger._is_current_root(trace):
             self._splunk_ao_logger.conclude(output=output, status_code=status_code, conclude_all=True)
 
     def log_node_tree(self, node: Node) -> None:

src/splunk_ao/handlers/openai_agents/handler.py

@@ -110,17 +110,7 @@ def _commit_trace(self, trace: Trace) -> None:
         self._splunk_ao_logger.conclude(output=self._last_output, status_code=self._last_status_code)
 
     def _conclude_current_trace_on_failure(self) -> None:
-        if self._owned_trace is None:
-            return
-
-        current_parent = self._splunk_ao_logger.current_parent()
-        if current_parent is None:
-            return
-
-        root = current_parent
-        while root._parent is not None:
-            root = root._parent
-        if root is self._owned_trace:
+        if self._splunk_ao_logger._is_current_root(self._owned_trace):
             self._splunk_ao_logger.conclude(output="", status_code=500, conclude_all=True)
 
     def _log_node_tree(self, node: Node, first_node: bool = False) -> None:

src/splunk_ao/logger/logger.py

@@ -368,7 +368,9 @@ def __init__(
                     "User must provide project_name or project_id to SplunkAOLogger, or set it as an environment variable."
                 )
             if self.experiment_id is None and self.agent_stream_name is None and self.agent_stream_id is None:
-                raise SplunkAOLoggerException("agent_stream or agent_stream_id is required to initialize SplunkAOLogger.")
+                raise SplunkAOLoggerException(
+                    "agent_stream or agent_stream_id is required to initialize SplunkAOLogger."
+                )
 
         if local_metrics:
             self.local_metrics = local_metrics
@@ -413,12 +415,20 @@ def _set_current_parent(self, parent: StepWithChildSpans | None) -> None:
         super()._set_current_parent(parent)
         self._sync_otel_context(parent)
 
-    def reset_parent_tracking(self) -> None:
-        """Clear proprietary and OTel tracking for the current request context."""
-        current_parent = self.current_parent()
-        root = current_parent
+    def _current_root(self) -> StepWithChildSpans | None:
+        """Return the root of the current proprietary parent chain."""
+        root = self.current_parent()
         while root is not None and root._parent is not None:
             root = root._parent
+        return root
+
+    def _is_current_root(self, trace: Trace | None) -> bool:
+        """Return whether trace owns the current proprietary parent chain."""
+        return trace is not None and self._current_root() is trace
+
+    def reset_parent_tracking(self) -> None:
+        """Clear proprietary and OTel tracking for the current request context."""
+        root = self._current_root()
 
         self._set_current_parent(None)
         if root is not None:

tests/test_attribute_mapping.py

@@ -400,6 +400,54 @@ def test_orchestration_preserves_schema_valid_parts_and_tool_calls() -> None:
     ]
 
 
+def test_orchestration_preserves_explicit_empty_message_parts() -> None:
+    # Given: role-bearing input and output messages with explicitly empty parts
+    span = WorkflowSpan(
+        name="empty-parts-workflow", input='{"role":"user","parts":[]}', output='{"role":"assistant","parts":[]}'
+    )
+
+    # When: the messages are mapped to canonical OTel attributes
+    attrs = build_span_attributes(span)
+
+    # Then: empty parts remain empty instead of becoming a text part containing "[]"
+    assert json.loads(attrs["gen_ai.input.messages"]) == [{"role": "user", "parts": []}]
+    assert json.loads(attrs["gen_ai.output.messages"]) == [
+        {"role": "assistant", "parts": [], "finish_reason": "unknown"}
+    ]
+
+
+def test_orchestration_treats_explicit_empty_parts_as_authoritative() -> None:
+    # Given: role-bearing messages with both explicit empty parts and legacy content
+    span = WorkflowSpan(
+        name="both-content-fields-workflow",
+        input='{"role":"user","parts":[],"content":"ignored input"}',
+        output='{"role":"assistant","parts":[],"content":"ignored output"}',
+    )
+
+    # When: the messages are mapped to canonical OTel attributes
+    attrs = build_span_attributes(span)
+
+    # Then: the explicit OTel parts field takes precedence over legacy content
+    assert json.loads(attrs["gen_ai.input.messages"]) == [{"role": "user", "parts": []}]
+    assert json.loads(attrs["gen_ai.output.messages"]) == [
+        {"role": "assistant", "parts": [], "finish_reason": "unknown"}
+    ]
+
+
+def test_orchestration_keeps_missing_parts_and_empty_content_distinct() -> None:
+    # Given: role-bearing messages with no parts field and explicit empty content
+    span = WorkflowSpan(
+        name="empty-content-workflow", input='{"role":"user","content":""}', output='{"role":"assistant","content":""}'
+    )
+
+    # When: the messages are mapped to canonical OTel attributes
+    attrs = build_span_attributes(span)
+
+    # Then: existing empty-content behavior remains a typed empty text part
+    assert json.loads(attrs["gen_ai.input.messages"]) == [_text_message("user", "")]
+    assert json.loads(attrs["gen_ai.output.messages"]) == [_text_message("assistant", "", finish_reason="unknown")]
+
+
 @pytest.mark.parametrize(
     "span",
     [
@@ -558,11 +606,7 @@ def test_control_mapping_omits_unpopulated_optional_fields() -> None:
 
 
 def test_control_mapping_accepts_schema_compatible_control_span() -> None:
-    source = ControlSpan(
-        name="guardrail",
-        output=ControlResult(action="observe", matched=True),
-        control_id=42,
-    )
+    source = ControlSpan(name="guardrail", output=ControlResult(action="observe", matched=True), control_id=42)
     alternate = SimpleNamespace(
         **{field_name: getattr(source, field_name) for field_name in type(source).model_fields}, model_extra={}
     )
@@ -601,8 +645,7 @@ def test_control_mapping_tolerates_span_without_control_fields() -> None:
 
 def test_control_mapping_exports_error_result_without_dropping_false() -> None:
     span = ControlSpan(
-        name="guardrail",
-        output=ControlResult(action="observe", matched=False, error_message="evaluator unavailable"),
+        name="guardrail", output=ControlResult(action="observe", matched=False, error_message="evaluator unavailable")
     )
 
     attrs = build_span_attributes(span)

tests/test_decorator_operation_ownership.py

@@ -86,6 +86,26 @@ def failing_operation() -> None:
     logger = splunk_ao_context.get_logger_instance()
     assert logger.current_parent() is None
     assert splunk_ao_context.get_current_trace() is None
+    assert (logger._sink.spans[-1].attributes or {})["splunk_ao.status_code"] == 500
+
+
+@pytest.mark.asyncio
+async def test_async_coroutine_exception_is_preserved_and_owned_trace_is_concluded(initialized_context: None) -> None:
+    # Given: a decorated async operation that raises an application exception
+    @log(span_type="workflow")
+    async def failing_operation() -> None:
+        await asyncio.sleep(0)
+        raise RuntimeError("async application failure")
+
+    # When: the operation is awaited
+    with pytest.raises(RuntimeError, match="async application failure"):
+        await failing_operation()
+
+    # Then: the original exception is re-raised and both telemetry contexts are released
+    logger = splunk_ao_context.get_logger_instance()
+    assert logger.current_parent() is None
+    assert splunk_ao_context.get_current_trace() is None
+    assert (logger._sink.spans[-1].attributes or {})["splunk_ao.status_code"] == 500
 
 
 def test_sync_generator_concludes_on_close_and_preserves_errors(initialized_context: None) -> None:

tests/test_exporter_diagnostics.py

@@ -21,6 +21,7 @@
     DiagnosticOTLPSpanExporter,
     ExportHealth,
     _ExportHealthTracker,
+    _positive_json_integer,
     _RejectionDetail,
 )
 from splunk_ao.exporter.sink import BatchConfig, build_span_sink
@@ -108,19 +109,70 @@ def test_positive_otlp_partial_success_returns_failure_without_retry() -> None:
     assert exporter._attempt_local.response is None
 
 
-@pytest.mark.parametrize("rejected_spans", [3, "3"])
-def test_positive_json_otlp_partial_success_returns_failure_without_backend_detail(rejected_spans: int | str) -> None:
+@pytest.mark.parametrize("rejected_spans", [3, "3", 3.0])
+def test_positive_json_otlp_partial_success_returns_failure_without_backend_detail(
+    rejected_spans: int | float | str,
+) -> None:
+    # Given: a successful OTLP JSON response with an unambiguous positive rejection count
     body = json.dumps(
         {"partialSuccess": {"rejectedSpans": rejected_spans, "errorMessage": "do not retain this backend detail"}}
     ).encode()
     exporter = diagnostic_exporter(FakeSession(response(body=body)))
 
-    assert exporter.export(()) == SpanExportResult.FAILURE
+    # When: the response is classified
+    result = exporter.export(())
+
+    # Then: each supported JSON representation produces the same bounded count without backend detail
+    assert result == SpanExportResult.FAILURE
     assert exporter.export_health.last_failure is not None
     assert "Rejected spans: 3" in exporter.export_health.last_failure.message
     assert "backend detail" not in exporter.export_health.last_failure.message
 
 
+@pytest.mark.parametrize(
+    "value",
+    [
+        True,
+        False,
+        0,
+        -1,
+        0.0,
+        -3.0,
+        3.5,
+        float(2**53),
+        1e30,
+        float("nan"),
+        float("inf"),
+        float("-inf"),
+        "",
+        "+3",
+        "-3",
+        "3.0",
+        "3e0",
+        {},
+        [],
+    ],
+)
+def test_invalid_json_rejected_span_counts_are_ignored(value: object) -> None:
+    # Given: a value that does not unambiguously represent a positive integer
+    # When: the acknowledgement count is parsed
+    result = _positive_json_integer(value)
+
+    # Then: the value is ignored without raising
+    assert result is None
+
+
+def test_max_safe_json_float_rejected_span_count_is_accepted() -> None:
+    # Given: the largest integer-valued float that JSON can represent unambiguously
+    value = float(2**53 - 1)
+
+    # When: the acknowledgement count is parsed
+    result = _positive_json_integer(value)
+
+    # Then: the exact safe integer value is retained
+    assert result == 2**53 - 1
+
+
 @pytest.mark.parametrize(
     ("body", "content_type"),
     [

tests/test_logger_otel_context.py

@@ -9,6 +9,7 @@
 
 from splunk_ao.logger import SplunkAOLogger
 from splunk_ao.logger.logger import _otel_context_state
+from splunk_ao.schema.logged import LoggedTrace
 
 
 @pytest.fixture(autouse=True)
@@ -56,6 +57,29 @@ def test_start_trace_assigns_and_activates_fresh_context(make_logger: Callable[[
     assert logger._otel_ids == {}
 
 
+def test_logger_identifies_current_root_ownership(make_logger: Callable[[], SplunkAOLogger]) -> None:
+    # Given: an empty logger and an unrelated proprietary trace
+    logger = make_logger()
+    unrelated_trace = LoggedTrace(input="unrelated")
+
+    # Then: absent inputs and an empty parent chain are never owned
+    assert logger._current_root() is None
+    assert logger._is_current_root(None) is False
+    assert logger._is_current_root(unrelated_trace) is False
+
+    # When: an owned root and nested current child are created
+    owned_trace = logger.start_trace(input="request", name="owned")
+    assert logger._current_root() is owned_trace
+    assert logger._is_current_root(owned_trace) is True
+    logger.add_workflow_span(input="nested", name="nested")
+
+    # Then: the root is discovered by identity and an unrelated trace is rejected
+    assert logger._current_root() is owned_trace
+    assert logger._is_current_root(owned_trace) is True
+    assert logger._is_current_root(unrelated_trace) is False
+    logger.conclude(output="done", conclude_all=True)
+
+
 def test_every_path1_step_gets_stable_ids_and_actual_parent(make_logger: Callable[[], SplunkAOLogger]) -> None:
     logger = make_logger()
     root = logger.start_trace(input="q")