diff --git a/CHANGELOG.md b/CHANGELOG.md index d8335195..d757ad35 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/src/splunk_ao/converter/attribute_mapping.py b/src/splunk_ao/converter/attribute_mapping.py index f753f818..cbcb9052 100644 --- a/src/splunk_ao/converter/attribute_mapping.py +++ b/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)} diff --git a/src/splunk_ao/decorator.py b/src/splunk_ao/decorator.py index 99dd26ec..3a0fdf6e 100644 --- a/src/splunk_ao/decorator.py +++ b/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: diff --git a/src/splunk_ao/exporter/diagnostics.py b/src/splunk_ao/exporter/diagnostics.py index 459a2ead..76422dad 100644 --- a/src/splunk_ao/exporter/diagnostics.py +++ b/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 diff --git a/src/splunk_ao/handlers/agent_control/bridge.py b/src/splunk_ao/handlers/agent_control/bridge.py index 74b2ed12..2e3099ed 100644 --- a/src/splunk_ao/handlers/agent_control/bridge.py +++ b/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)) diff --git a/src/splunk_ao/handlers/base_handler.py b/src/splunk_ao/handlers/base_handler.py index c095e717..d94863f0 100644 --- a/src/splunk_ao/handlers/base_handler.py +++ b/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: diff --git a/src/splunk_ao/handlers/openai_agents/handler.py b/src/splunk_ao/handlers/openai_agents/handler.py index 6826a880..d5cb4741 100644 --- a/src/splunk_ao/handlers/openai_agents/handler.py +++ b/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: diff --git a/src/splunk_ao/logger/logger.py b/src/splunk_ao/logger/logger.py index 8ebd8937..e571c183 100644 --- a/src/splunk_ao/logger/logger.py +++ b/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: diff --git a/tests/test_attribute_mapping.py b/tests/test_attribute_mapping.py index 5fff65ec..a05a9c0a 100644 --- a/tests/test_attribute_mapping.py +++ b/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) diff --git a/tests/test_decorator_operation_ownership.py b/tests/test_decorator_operation_ownership.py index 55308865..8f5ad3e0 100644 --- a/tests/test_decorator_operation_ownership.py +++ b/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: diff --git a/tests/test_exporter_diagnostics.py b/tests/test_exporter_diagnostics.py index ef8e64da..e0204444 100644 --- a/tests/test_exporter_diagnostics.py +++ b/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"), [ diff --git a/tests/test_logger_otel_context.py b/tests/test_logger_otel_context.py index a9481b26..75eefbc3 100644 --- a/tests/test_logger_otel_context.py +++ b/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")