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,9 @@ 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:
+    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

@@ -903,12 +903,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
@@ -247,6 +248,9 @@ 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():
+        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/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,6 +415,18 @@ def _set_current_parent(self, parent: StepWithChildSpans | None) -> None:
         super()._set_current_parent(parent)
         self._sync_otel_context(parent)
 
+    def _is_current_root(self, trace: Trace | None) -> bool:
+        """Return whether trace owns the current proprietary parent chain."""
+        if trace is None:
+            return False
+
+        root = self.current_parent()
+        if root is None:
+            return False
+        while root._parent is not None:
+            root = root._parent
+        return root is trace
+
     def reset_parent_tracking(self) -> None:
         """Clear proprietary and OTel tracking for the current request context."""
         current_parent = self.current_parent()

tests/test_attribute_mapping.py

@@ -400,6 +400,36 @@ 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_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 +588,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 +627,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

@@ -5,6 +5,7 @@
 import pytest
 
 from splunk_ao import log, splunk_ao_context
+from splunk_ao.schema.logged import LoggedTrace
 from tests.testutils.setup import setup_mock_logstreams_client, setup_mock_projects_client, setup_mock_traces_client
 
 
@@ -88,6 +89,44 @@ def failing_operation() -> None:
     assert splunk_ao_context.get_current_trace() is None
 
 
+@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
+
+
+def test_logger_identifies_current_root_ownership(initialized_context: None) -> None:
+    # Given: an empty logger and an unrelated proprietary trace
+    logger = splunk_ao_context.get_logger_instance()
+    unrelated_trace = LoggedTrace(input="unrelated")
+
+    # Then: absent inputs and an empty parent chain are never owned
+    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._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._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_sync_generator_concludes_on_close_and_preserves_errors(initialized_context: None) -> None:
     @log(span_type="workflow")
     def stream(fail: bool = False):

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,57 @@ 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("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
+
+
 @pytest.mark.parametrize(
     ("body", "content_type"),
     [