From a3fdcd18c8788848e09e92d2e95e86b70d67bf2f Mon Sep 17 00:00:00 2001 From: pradystar Date: Fri, 23 Oct 2026 08:06:16 -0700 Subject: [PATCH 1/2] feat(logger): preserve manual trace declarations in OTLP spans - Capture immutable trace fields on direct operation spans - Keep automatic handler envelopes unmarked - Reject known local trace redeclarations - Deprecate start_trace timing and external ID arguments - Add regression coverage and document backend compatibility --- AGENTS.md | 6 + ARCHITECTURE.md | 16 ++ CHANGELOG.md | 11 + README.md | 27 ++ src/splunk_ao/converter/span_converter.py | 8 +- src/splunk_ao/decorator.py | 10 +- src/splunk_ao/handlers/base_handler.py | 2 +- .../handlers/openai_agents/handler.py | 2 +- src/splunk_ao/logger/logger.py | 97 ++++++- src/splunk_ao/logger/trace_metadata.py | 39 +++ src/splunk_ao/openai/__init__.py | 2 +- src/splunk_ao/schema/logged.py | 4 +- tests/conftest.py | 3 + tests/test_base_handler.py | 1 + tests/test_logger_otel_context.py | 4 +- tests/test_logger_otel_egress.py | 244 ++++++++++++++++++ 16 files changed, 458 insertions(+), 18 deletions(-) create mode 100644 src/splunk_ao/logger/trace_metadata.py diff --git a/AGENTS.md b/AGENTS.md index 6f2bd531..2415e1ab 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -133,6 +133,12 @@ CI supports Python 3.11–3.14; root CI also spans Linux, macOS, and Windows. - Handler/decorator/OpenAI/ADK telemetry uses the internal logged-step path and converts completed steps to immutable OTel spans. The internal trace envelope is never exported as a span. +- Manual `start_trace()` freezes supplied fields privately; only direct real operation children carry `splunk_ao.trace.*`. + Keep operation input separate. Internal integrations use unmarked `_start_trace_envelope()` bookkeeping; low-level + `add_trace()` remains unmarked. Public `created_at`, `duration_ns`, and `external_id` are deprecated and ignored. +- Preserve bounded content-free execution-local trace claims across logger conclusion/termination. Reject known local + redeclarations; automatic decorator continuations must neither redeclare nor replay content. Do not add ownership + propagation or a global trace registry. Tests must isolate the claim ContextVar together with OTel/session context. - `start_splunk_ao_span()` is SDK-native OTel. `add_splunk_ao_span_processor()` and A2A instrument caller-owned OTel. `configure_distributed_tracing()` is the combined automatic HTTP setup: it creates or accepts an application-owned provider, registers Splunk AO export once per provider, configures supported upstream instrumentors, and returns the diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index cda7b73c..2ac161db 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -67,6 +67,22 @@ Path 3: external OTel/OpenInference + A2A internal trace/step model. Stable OTel context is associated with actual operations. Completed steps are converted to OTel spans; the `LoggedTrace` envelope remains an internal lifecycle container and must not become a wire span. +Manual `start_trace()` and independent top-level decorators freeze supplied trace fields into a private immutable +snapshot, excluded from legacy hook model dumps. Actual operations directly parented by that envelope carry identical +`splunk_ao.trace.*` attributes; descendants do not. Conversion adds them before constructing `ReadableSpan`, without +changing operation input, IDs, timing, parentage, or sampling. Snapshots are released with their envelope. +Backend trace projection requires compatible new Go OTel ingestion; no trace CRUD request is added. + +Internal integration envelopes use `_start_trace_envelope()` without a declaration. The low-level `add_trace()` and +single-LLM convenience path remain unmarked. Deprecated public `start_trace()` timing and external-ID arguments are +ignored; private integration timing and actual operation timing remain supported. + +One content-free trace-ID claim per execution context detects known local redeclarations across logger instances and +local conclusion. Automatic decorator continuations open unmarked envelopes, without replaying the original snapshot. +The claim is not cleared by a logger's conclusion or termination. Independent calls without a remaining upstream parent +create fresh traces. This is bounded local detection, not global ownership: origins declare once and downstream services +continue spans. No baggage, propagation, global trace registry, provider replacement, or sampling change is involved. + Do not bypass this path when extending an existing handler. It centralizes lifecycle behavior, content conversion, session handling, and compatibility hooks. Conversion changes must cover workflow, agent, LLM, tool, retriever, and Agent Control kinds that they affect. diff --git a/CHANGELOG.md b/CHANGELOG.md index 72556781..b299134c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Manual `start_trace()` fields are frozen and carried as dedicated `splunk_ao.trace.*` attributes on direct real + operation spans, separate from operation input. Requires compatible new Go OTel ingestion for backend projection. + Top-level decorators include mapped metadata/tags. Known local same-trace redeclarations are rejected; automatic + integration/decorator continuation envelopes remain unmarked. No artificial span or CRUD request is introduced. + - `create_decision_evaluator()` (also `Evaluators.create_decision_evaluator`) and `DecisionEvaluator` create and read decision-model evaluators: one typed question per evaluator, answered with probabilities. Supports `boolean_probability`, `categorical` and `discrete` output types; when `output_type` is omitted it is inferred @@ -17,6 +22,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Scorer requests now send `X-Galileo-Scorer-Capabilities: decision`, and `ScorerTypes` and `ModelType` include `decision`, so decision scorers appear in list and get responses and decode. +### Deprecated + +- Public `start_trace(created_at=..., duration_ns=..., external_id=...)` accepts but ignores these values and emits + `DeprecationWarning`, including hook-backed callers. Actual operation timing APIs are unchanged; use metadata for + business identifiers. Declare trace fields only at the origin, not again in downstream distributed services. + ### Fixed - `SplunkAOTracingProcessor` (OpenAI Agents) now records Responses API LLM spans, the Agents SDK diff --git a/README.md b/README.md index 0199c87c..47b9ed5b 100644 --- a/README.md +++ b/README.md @@ -282,6 +282,33 @@ for normal scheduled export. `terminate()` drains completed spans and shuts down resources owned by `SplunkAOLogger`; it does not turn unfinished manual spans into completed spans. +`start_trace()` opens an internal envelope, not an artificial root span or a +backend CRUD resource. It freezes supplied trace `name`, `input`, +`redacted_input`, `metadata`, `tags`, `dataset_input`, `dataset_output`, and +`dataset_metadata`. Real operations directly below the envelope carry these +values as `splunk_ao.trace.*` attributes; nested operations retain only their +own fields. Trace input never replaces an operation's input. Explicit empty +values are preserved; later edits to supplied collections do not change the +declaration. Backend trace-field projection requires compatible new Go-based +OTel ingestion support; older ingestion routes are not supported for this feature. + +Declare fields once in the originating application. A duplicate active call, +or a redeclaration of the same OTel trace known to the local execution context, +raises `ValueError`. `conclude()` ends local bookkeeping, not a distributed +trace. Downstream services should continue operations without redeclaring trace +fields. Automatic integration envelopes do not declare fields; independent +top-level decorators declare once, then use unmarked envelopes for known local +continuations. No ownership marker is propagated, so the SDK cannot detect +redeclarations in independent services or execution contexts. Propagated worker +contexts can share a trace ID even across threads. + +The `start_trace()` arguments `created_at`, `duration_ns`, and `external_id` +are deprecated and ignored, including with an `ingestion_hook`. Actual operation +timing APIs are unchanged. The backend uses the OTel trace ID for correlation; +put business identifiers in metadata instead. No artificial span, sampling +override, or separate network call is added. Without an exported real operation, +the declaration is not delivered. + #### Using Agent Observability context with Agent Control If you use Agent Control hosted by Splunk, initialize Agent Control with the diff --git a/src/splunk_ao/converter/span_converter.py b/src/splunk_ao/converter/span_converter.py index d041a482..f8893aa7 100644 --- a/src/splunk_ao/converter/span_converter.py +++ b/src/splunk_ao/converter/span_converter.py @@ -3,12 +3,14 @@ from __future__ import annotations import time +from collections.abc import Mapping from datetime import UTC, datetime from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import ReadableSpan from opentelemetry.sdk.util.instrumentation import InstrumentationScope from opentelemetry.trace import SpanContext, SpanKind, Status, StatusCode +from opentelemetry.util.types import AttributeValue from galileo_core.schemas.logging.step import BaseStep, StepType from splunk_ao.converter.attribute_mapping import build_span_attributes @@ -85,17 +87,21 @@ def convert_span( session_id: str | None, resource: Resource, end_time_ns: int | None = None, + trace_attributes: Mapping[str, AttributeValue] | None = None, ) -> ReadableSpan: """Build a readable span using the supplied identity and resource.""" step_type = _step_type(span) start_time_ns = _to_unix_ns(span.created_at) + attributes = build_span_attributes(span, session_id) + if trace_attributes: + attributes.update(trace_attributes) return ReadableSpan( name=_span_name(span, step_type), context=span_context, parent=parent_span_context, resource=resource, - attributes=build_span_attributes(span, session_id), + attributes=attributes, events=(), links=(), kind=_KIND_BY_STEP_TYPE[step_type], diff --git a/src/splunk_ao/decorator.py b/src/splunk_ao/decorator.py index 6188e839..ff64ab1f 100644 --- a/src/splunk_ao/decorator.py +++ b/src/splunk_ao/decorator.py @@ -61,6 +61,7 @@ def call_llm(prompt, temperature=0.7): from galileo_core.schemas.logging.trace import Trace from splunk_ao.constants import LoggerModeType from splunk_ao.logger import SplunkAOLogger +from splunk_ao.logger.trace_metadata import has_trace_declaration from splunk_ao.schema.content_blocks import is_content_block_list from splunk_ao.schema.datasets import DatasetRecord from splunk_ao.schema.metrics import LocalMetricConfig @@ -778,9 +779,16 @@ def _prepare_call( if not isinstance(trace, Trace): raise RuntimeError("Active Splunk AO operation does not have a trace root") else: - trace = client_instance.start_trace( + # Independent calls declare once; known distributed continuations + # need only local bookkeeping, never a second trace declaration. + start_trace = ( + client_instance._start_trace_envelope if has_trace_declaration() else client_instance.start_trace + ) + trace = start_trace( input=input_, name=name, + metadata=span_params.get("metadata"), + tags=span_params.get("tags"), # TODO: add dataset_row_id dataset_input=dataset_record.input if dataset_record else None, dataset_output=dataset_record.output if dataset_record else None, diff --git a/src/splunk_ao/handlers/base_handler.py b/src/splunk_ao/handlers/base_handler.py index 589939ab..864facfe 100644 --- a/src/splunk_ao/handlers/base_handler.py +++ b/src/splunk_ao/handlers/base_handler.py @@ -252,7 +252,7 @@ def _start_owned_root(self, root_node: Node) -> None: return if self._owned_trace is None: - self._owned_trace = self._splunk_ao_logger.start_trace( + self._owned_trace = self._splunk_ao_logger._start_trace_envelope( input=SplunkAOLogger._coerce_output(root_node.span_params.get("input", "")), name=root_node.span_params.get("name"), metadata=root_node.span_params.get("metadata"), diff --git a/src/splunk_ao/handlers/openai_agents/handler.py b/src/splunk_ao/handlers/openai_agents/handler.py index 37bc33da..a5ecf80b 100644 --- a/src/splunk_ao/handlers/openai_agents/handler.py +++ b/src/splunk_ao/handlers/openai_agents/handler.py @@ -108,7 +108,7 @@ def on_trace_start(self, trace: Trace) -> None: self._trace_states[trace_id] = state if state.caller_parent is None: try: - state.owned_trace = self._splunk_ao_logger.start_trace( + state.owned_trace = self._splunk_ao_logger._start_trace_envelope( input="Agent Workflow", name="Trace", created_at=datetime.fromisoformat(node.span_params["start_time_iso"]), diff --git a/src/splunk_ao/logger/logger.py b/src/splunk_ao/logger/logger.py index 1c4241eb..263fb86d 100644 --- a/src/splunk_ao/logger/logger.py +++ b/src/splunk_ao/logger/logger.py @@ -6,6 +6,7 @@ import logging import time import uuid +import warnings from collections.abc import Callable from contextvars import ContextVar, Token from dataclasses import dataclass @@ -49,6 +50,7 @@ ) from splunk_ao.exporter.diagnostics import get_export_health from splunk_ao.logger.control import ControlAppliesTo, ControlCheckStage, ControlResult +from splunk_ao.logger.trace_metadata import capture_trace_attributes, claim_trace, has_trace_declaration from splunk_ao.projects import Projects from splunk_ao.schema.content_blocks import ( DataContentBlock, @@ -584,6 +586,11 @@ def _emit_and_release(self, finished_step: BaseStep) -> None: parent_span_context=self._export_parent_context(finished_step, ids), session_id=ids.session_id, resource=self._resource, + trace_attributes=( + dict(finished_step._parent._trace_attributes) + if isinstance(finished_step._parent, LoggedTrace) + else None + ), ) ) except Exception: @@ -606,11 +613,15 @@ def _complete_step(self, finished_step: BaseStep) -> None: """Coordinate completion for hook or OTLP egress without double dispatch.""" if self._ingestion_hook: self._release_otel_context(finished_step) + if isinstance(finished_step, LoggedTrace): + finished_step._trace_attributes = () return self._emit_pending_descendants(finished_step) self._emit_and_release(finished_step) if isinstance(finished_step, Trace): + if isinstance(finished_step, LoggedTrace): + finished_step._trace_attributes = () self.traces = [trace for trace in getattr(self, "traces", []) if trace is not finished_step] def _add_completed_leaf(self, span: StepT, parent: BaseStep | None = None) -> StepT: @@ -1006,8 +1017,13 @@ def start_trace( external_id: str | None = None, ) -> LoggedTrace: """ - Create a new trace and add it to the list of traces. - Once this trace is complete, you can close it out by calling conclude(). + Declare trace fields and open an internal instrumentation envelope. + + No span or backend resource is created by this call alone. Completed real + operations directly below this envelope carry a frozen declaration using + ``splunk_ao.trace.*`` attributes, separate from their operation input. + Call once at the originating application, then close the local envelope + with ``conclude()``. Distributed continuations must not redeclare fields. Parameters ---------- @@ -1028,9 +1044,9 @@ def start_trace( Name of the trace. Example: "weather_query_trace", "customer_support_session" duration_ns: Optional[int] - Duration of the trace in nanoseconds. + Deprecated and ignored. Actual operations supply their own duration. created_at: Optional[datetime] - Timestamp of the trace's creation. + Deprecated and ignored. Actual operations supply their own timestamps. metadata: Optional[dict[str, MetadataValue]] Metadata associated with this trace. Expected format: `{"key1": "value1", "enabled": True, "count": 42}` @@ -1048,25 +1064,77 @@ def start_trace( Expected format: `{"key1": "value1", "enabled": True, "count": 42}` Accepted value types: str, bool, int, float, None (auto-converted to strings). external_id: Optional[str] - External ID for this trace to connect to external systems. - Expected format: Unique identifier string. + Deprecated and ignored. The backend correlates using the OTel trace ID. + Put business identifiers in metadata instead. Returns ------- LoggedTrace The created trace. """ - # Validate and normalize input types (dict→JSON, list[dict]→JSON or content blocks) + for argument, value in (("created_at", created_at), ("duration_ns", duration_ns), ("external_id", external_id)): + if value is not None: + warnings.warn(f"start_trace() {argument} is deprecated and ignored.", DeprecationWarning, stacklevel=4) + return self._start_trace_envelope( + input=input, + redacted_input=redacted_input, + name=name, + metadata=metadata, + tags=tags, + dataset_input=dataset_input, + dataset_output=dataset_output, + dataset_metadata=dataset_metadata, + _declare=True, + ) + + @nop_sync + @warn_catch_exception() + def _start_trace_envelope( + self, + input: str | TextOrContentBlocks | dict | list[dict[str, Any]], + redacted_input: str | TextOrContentBlocks | dict | list[dict[str, Any]] | None = None, + name: str | None = None, + duration_ns: int | None = None, + created_at: datetime | None = None, + metadata: dict[str, MetadataValue] | None = None, + tags: list[str] | None = None, + dataset_input: str | None = None, + dataset_output: str | None = None, + dataset_metadata: dict[str, MetadataValue] | None = None, + *, + _declare: bool = False, + ) -> LoggedTrace: + """Open bookkeeping state; only public manual/decorator calls declare fields.""" + if self.current_parent() is not None: + raise ValueError("You must conclude the existing trace before adding a new one.") + if _declare and has_trace_declaration(): + raise ValueError("Trace fields have already been declared in this execution context.") input = SplunkAOLogger._coerce_trace_input("input", input) redacted_input = SplunkAOLogger._coerce_trace_input("redacted_input", redacted_input) # Auto-convert non-string metadata values to strings # Note: Must use class name, not self, because DecorateAllMethods removes @staticmethod - if metadata: + if metadata is not None: metadata = {k: SplunkAOLogger._convert_metadata_value(v) for k, v in metadata.items()} - if dataset_metadata: + if dataset_metadata is not None: dataset_metadata = {k: SplunkAOLogger._convert_metadata_value(v) for k, v in dataset_metadata.items()} + attributes = ( + capture_trace_attributes( + **{ + "input": input, + "redacted_input": redacted_input, + "name": name, + "metadata": metadata, + "tags": tags, + "dataset.input": dataset_input, + "dataset.output": dataset_output, + "dataset.metadata": dataset_metadata, + } + ) + if _declare + else () + ) kwargs = { "input": input, "redacted_input": redacted_input, @@ -1078,10 +1146,14 @@ def start_trace( "dataset_input": dataset_input, "dataset_output": dataset_output, "dataset_metadata": dataset_metadata, - "external_id": external_id, "id": uuid.uuid4(), } - return self.add_trace(**kwargs) + trace = self.add_trace(**kwargs) + trace._trace_attributes = attributes + ids = self._otel_ids.get(trace.id) + if _declare and ids is not None: + claim_trace(ids.span_context.trace_id) + return trace @nop_sync @warn_catch_exception(exceptions=(Exception,)) @@ -2110,6 +2182,9 @@ def terminate(self) -> None: self._set_current_parent(None) self._otel_ids.clear() self._pending_otel_steps.clear() + for trace in self.traces: + if isinstance(trace, LoggedTrace): + trace._trace_attributes = () self.traces = [] try: diff --git a/src/splunk_ao/logger/trace_metadata.py b/src/splunk_ao/logger/trace_metadata.py new file mode 100644 index 00000000..ed8a7004 --- /dev/null +++ b/src/splunk_ao/logger/trace_metadata.py @@ -0,0 +1,39 @@ +"""Private immutable trace declarations; never propagated as OTel context content.""" + +import json +from contextvars import ContextVar +from typing import Any + +from opentelemetry import trace + +TraceAttributes = tuple[tuple[str, str | int | tuple[str, ...]], ...] + +# One content-free claim per execution context, not a process-wide trace registry. +_declared_trace_id: ContextVar[int | None] = ContextVar("splunk_ao_declared_trace_id", default=None) + + +def has_trace_declaration() -> bool: + """Whether the active OTel trace was declared in this execution context.""" + current = trace.get_current_span().get_span_context() + return current.is_valid and current.trace_id == _declared_trace_id.get() + + +def claim_trace(trace_id: int) -> None: + """Remember only the current local declaration's identity.""" + _declared_trace_id.set(trace_id) + + +def capture_trace_attributes(**values: Any) -> TraceAttributes: + """Freeze supplied normalized values without leaking mutable model references.""" + attributes: list[tuple[str, str | int | tuple[str, ...]]] = [("splunk_ao.trace.schema_version", 1)] + for field, value in values.items(): + if value is None: + continue + if field in ("input", "redacted_input", "metadata", "dataset.metadata"): + if isinstance(value, list): + value = [block.model_dump(mode="json", exclude_none=True) for block in value] + value = json.dumps(value) + elif field == "tags": + value = tuple(value) + attributes.append((f"splunk_ao.trace.{field}", value)) + return tuple(attributes) diff --git a/src/splunk_ao/openai/__init__.py b/src/splunk_ao/openai/__init__.py index e47e9edc..ba79a15f 100644 --- a/src/splunk_ao/openai/__init__.py +++ b/src/splunk_ao/openai/__init__.py @@ -152,7 +152,7 @@ def _wrap( # Serialize with "messages" wrapper for UI compatibility trace_input = {"messages": [msg.model_dump(exclude_none=True) for msg in trace_input_messages]} - splunk_ao_logger.start_trace(input=serialize_to_str(trace_input), name=input_data.name) + splunk_ao_logger._start_trace_envelope(input=serialize_to_str(trace_input), name=input_data.name) should_complete_trace = True try: diff --git a/src/splunk_ao/schema/logged.py b/src/splunk_ao/schema/logged.py index 01565318..5882796e 100644 --- a/src/splunk_ao/schema/logged.py +++ b/src/splunk_ao/schema/logged.py @@ -9,7 +9,7 @@ from json import dumps from typing import Annotated, Any -from pydantic import Field +from pydantic import Field, PrivateAttr from galileo_core.schemas.logging.llm import Message, MessageRole from galileo_core.schemas.logging.span import ( @@ -26,6 +26,7 @@ from galileo_core.schemas.logging.trace import Trace from galileo_core.schemas.shared.document import Document from splunk_ao.logger.control import ControlSpan +from splunk_ao.logger.trace_metadata import TraceAttributes from splunk_ao.schema.content_blocks import IngestContentBlock, IngestMessageContent from splunk_ao.schema.message import LoggedMessage @@ -51,6 +52,7 @@ class LoggedTrace(Trace): output: TextOrContentBlocks | None = _OUTPUT_FIELD redacted_output: TextOrContentBlocks | None = _REDACTED_OUTPUT_FIELD spans: list["LoggedSpan"] = Field(default_factory=list) + _trace_attributes: TraceAttributes = PrivateAttr(default=()) class LoggedWorkflowSpan(WorkflowSpan): diff --git a/tests/conftest.py b/tests/conftest.py index 3f046681..7f47e883 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -63,6 +63,7 @@ from splunk_ao.config import SplunkAOConfig from splunk_ao.configuration import _CONFIGURATION_KEYS, Configuration from splunk_ao.decorator import _mode_context, _session_id_context +from splunk_ao.logger.trace_metadata import _declared_trace_id from splunk_ao.resources.models import DatasetContent, DatasetRow, DatasetRowValuesDict from splunk_ao.resources.models.messages_list_item import MessagesListItem from splunk_ao.utils.singleton import SplunkAOLoggerSingleton @@ -157,11 +158,13 @@ def _clear_otel_test_context() -> None: @pytest.fixture(autouse=True) def reset_otel_test_context() -> Generator[None, None, None]: + claim_token = _declared_trace_id.set(None) _clear_otel_test_context() _session_id_context.set(None) yield _clear_otel_test_context() _session_id_context.set(None) + _declared_trace_id.reset(claim_token) @pytest.fixture(autouse=True) diff --git a/tests/test_base_handler.py b/tests/test_base_handler.py index b4dab0cc..41386400 100644 --- a/tests/test_base_handler.py +++ b/tests/test_base_handler.py @@ -131,6 +131,7 @@ def test_normal_otel_child_enqueues_at_callback_end_without_flush() -> None: child_span, root_span = sink.spans assert child_span.parent == root_span.context assert root_span.parent is None + assert not any(key.startswith("splunk_ao.trace.") for span in sink.spans for key in span.attributes) assert [child.name for child in root_step.spans] == ["child"] assert sink.force_flush_calls == 0 finally: diff --git a/tests/test_logger_otel_context.py b/tests/test_logger_otel_context.py index 62a84d35..0f5deefa 100644 --- a/tests/test_logger_otel_context.py +++ b/tests/test_logger_otel_context.py @@ -389,7 +389,9 @@ def test_interleaved_logger_reset_preserves_current_logger(make_logger: Callable logger_a = make_logger() logger_b = make_logger() logger_a.start_trace(input="a") - root_b = logger_b.start_trace(input="b") + with pytest.raises(ValueError, match=r"already.*declared"): + logger_b.start_trace(input="b") + root_b = logger_b._start_trace_envelope(input="b") workflow_b = logger_b.add_workflow_span(input="workflow-b") workflow_b_context = logger_b._otel_ids[workflow_b.id].span_context diff --git a/tests/test_logger_otel_egress.py b/tests/test_logger_otel_egress.py index f5ed7b74..61950f9d 100644 --- a/tests/test_logger_otel_egress.py +++ b/tests/test_logger_otel_egress.py @@ -1,6 +1,8 @@ import asyncio import json from collections.abc import Generator +from contextvars import copy_context +from datetime import UTC, datetime from unittest.mock import MagicMock, Mock, patch import pytest @@ -14,6 +16,7 @@ from splunk_ao.logger import SplunkAOLogger from splunk_ao.openai import _wrap from splunk_ao.openai.models import OpenAiModuleDefinition +from splunk_ao.schema.content_blocks import TextContentBlock from splunk_ao.shared.exceptions import MissingConfigurationError from splunk_ao.utils.singleton import SplunkAOLoggerSingleton @@ -66,6 +69,247 @@ def configure_o11y(monkeypatch: pytest.MonkeyPatch, *, ingest_token: bool = True monkeypatch.delenv("SPLUNK_AO_O11Y_TOKEN", raising=False) +def trace_attributes(span: ReadableSpan) -> dict: + return {key: value for key, value in span.attributes.items() if key.startswith("splunk_ao.trace.")} + + +def test_trace_declaration_is_frozen_and_repeated_only_on_direct_siblings( + otlp_logger: SplunkAOLogger, recording_sink: RecordingSink +) -> None: + # Given: an explicit trace declaration, separate from each operation's input + metadata = {"count": 3, "enabled": True, "missing": None} + tags = ["initial"] + dataset_metadata = {"row": 1} + envelope = otlp_logger.start_trace( + input='{"text": "not content blocks"}', + redacted_input="safe", + name="support", + metadata=metadata, + tags=tags, + dataset_input="question", + dataset_output="reference", + dataset_metadata=dataset_metadata, + ) + metadata["count"] = 9 + tags.append("later") + dataset_metadata["row"] = 2 + envelope.input = "changed envelope" + envelope.name = "changed name" + + # When: direct siblings finish, including an enclosing operation with a child + otlp_logger.add_llm_span(input="actual prompt", output="answer", model="model") + otlp_logger.add_workflow_span(input="actual workflow input", name="workflow") + otlp_logger.add_llm_span(input="nested prompt", output="answer", model="model") + otlp_logger.conclude(output="done") + otlp_logger.conclude() + + # Then: only direct operations carry the same original snapshot + first, nested, workflow = recording_sink.spans + expected = { + "splunk_ao.trace.schema_version": 1, + "splunk_ao.trace.input": json.dumps('{"text": "not content blocks"}'), + "splunk_ao.trace.redacted_input": json.dumps("safe"), + "splunk_ao.trace.name": "support", + "splunk_ao.trace.metadata": json.dumps({"count": "3", "enabled": "True", "missing": "None"}), + "splunk_ao.trace.tags": ("initial",), + "splunk_ao.trace.dataset.input": "question", + "splunk_ao.trace.dataset.output": "reference", + "splunk_ao.trace.dataset.metadata": json.dumps({"row": "1"}), + } + assert trace_attributes(first) == trace_attributes(workflow) == expected + assert envelope._trace_attributes == () + assert trace_attributes(nested) == {} + assert json.loads(first.attributes["gen_ai.input.messages"])[0]["parts"][0]["content"] == "actual prompt" + assert workflow.name == "invoke_workflow workflow" + assert workflow.parent is first.parent is None + assert nested.parent == workflow.context + + +@pytest.mark.parametrize("explicit_empty", [False, True]) +def test_trace_declaration_preserves_presence( + otlp_logger: SplunkAOLogger, recording_sink: RecordingSink, explicit_empty: bool +) -> None: + # Given/When: omitted fields or explicitly empty values + kwargs = dict(name="", metadata={}, tags=[], dataset_input="", dataset_output="", dataset_metadata={}) + otlp_logger.start_trace(input="", **(kwargs if explicit_empty else {})) + otlp_logger.add_llm_span(input="prompt", output="answer", model="model") + attributes = trace_attributes(recording_sink.spans[0]) + + # Then: model defaults never manufacture declared fields + assert attributes.pop("splunk_ao.trace.schema_version") == 1 + assert attributes.pop("splunk_ao.trace.input") == json.dumps("") + assert attributes == ( + { + "splunk_ao.trace.name": "", + "splunk_ao.trace.metadata": "{}", + "splunk_ao.trace.tags": (), + "splunk_ao.trace.dataset.input": "", + "splunk_ao.trace.dataset.output": "", + "splunk_ao.trace.dataset.metadata": "{}", + } + if explicit_empty + else {} + ) + + +def test_active_trace_redeclaration_does_not_mutate_state( + otlp_logger: SplunkAOLogger, recording_sink: RecordingSink +) -> None: + # Given: an active declaration + envelope = otlp_logger.start_trace(input="original", name="original") + identity = dict(otlp_logger._otel_ids) + # When: a second declaration is attempted + with pytest.raises(ValueError, match="conclude"): + otlp_logger.start_trace(input="replacement", name="replacement") + # Then: original lifecycle and declaration remain intact + assert otlp_logger.current_parent() is envelope + assert otlp_logger._otel_ids == identity + otlp_logger.add_llm_span(input="prompt", output="answer", model="model") + assert recording_sink.spans[0].attributes["splunk_ao.trace.name"] == "original" + + +@pytest.mark.parametrize("sampled", [False, True]) +def test_trace_declaration_rejects_known_continuation_across_loggers( + otlp_logger: SplunkAOLogger, recording_sink: RecordingSink, sampled: bool +) -> None: + # Given: a declaration beneath an upstream distributed parent + upstream = trace.SpanContext(0x12345, 0x6789, True, trace.TraceFlags(int(sampled)), trace.TraceState()) + token = context.attach(trace.set_span_in_context(trace.NonRecordingSpan(upstream))) + other = SplunkAOLogger(project_id="project-id", agent_stream_id="log-stream-id", _sink=recording_sink) + try: + otlp_logger.start_trace(input="original") + otlp_logger.add_llm_span(input="prompt", output="answer", model="model") + otlp_logger.conclude() + # When: another logger tries to redefine that same trace after local conclusion + with pytest.raises(ValueError, match=r"already.*declared"): + other.start_trace(input="replacement") + assert other.current_parent() is None + # Then: automatic bookkeeping continues without replaying the declaration + other._start_trace_envelope(input="automatic") + other.add_llm_span(input="continued", output="answer", model="model") + other.conclude() + assert trace.get_current_span().get_span_context() == upstream + assert len(recording_sink.spans) == (2 if sampled else 0) + if sampled: + assert all(span.context.trace_id == upstream.trace_id for span in recording_sink.spans) + assert all(span.parent == upstream for span in recording_sink.spans) + assert trace_attributes(recording_sink.spans[-1]) == {} + finally: + other.terminate() + context.detach(token) + + +def test_deprecated_start_trace_arguments_are_ignored(otlp_logger: SplunkAOLogger) -> None: + # Given/When: deprecated envelope identity and timing arguments + old_time = datetime(2000, 1, 1, tzinfo=UTC) + with pytest.warns(DeprecationWarning) as warnings: + envelope = otlp_logger.start_trace( + input="question", created_at=old_time, duration_ns=42, external_id="business-id" + ) + # Then: no user-supplied timing or external identity enters the envelope + assert len(warnings) == 3 + assert envelope.created_at != old_time + assert envelope.metrics.duration_ns is None + assert envelope.external_id != "business-id" + assert "_trace_attributes" not in envelope.model_dump() + + +def test_content_block_declaration_is_serialized_before_caller_mutation( + otlp_logger: SplunkAOLogger, recording_sink: RecordingSink +) -> None: + # Given: a mutable multimodal input model + block = TextContentBlock(text="original") + otlp_logger.start_trace(input=[block]) + # When: the caller edits the model before an operation completes + block.text = "edited" + otlp_logger.add_llm_span(input="actual", output="answer", model="model") + # Then: declaration encoding is stable and survives export normalization + exported = copy_span_for_export(recording_sink.spans[0], normalize_attributes=True) + assert json.loads(exported.attributes["splunk_ao.trace.input"]) == [{"type": "text", "text": "original"}] + + +def test_independent_traces_and_execution_contexts_do_not_share_declarations(recording_sink: RecordingSink) -> None: + # Given: two execution contexts copied before either originates a declaration + first_context, second_context = copy_context(), copy_context() + + def operation(name: str) -> int: + logger = SplunkAOLogger(project_id="project-id", agent_stream_id="log-stream-id", _sink=recording_sink) + try: + logger.start_trace(input=name, name=name) + logger.add_llm_span(input="actual", output="answer", model="model") + logger.conclude() + return recording_sink.spans[-1].context.trace_id + finally: + logger.terminate() + + # When: independent calls run in those contexts, and again without an upstream parent + first_id = first_context.run(operation, "first") + second_id = second_context.run(operation, "second") + third_id = first_context.run(operation, "third") + # Then: identities and declarations remain independent + assert len({first_id, second_id, third_id}) == 3 + assert [span.attributes["splunk_ao.trace.name"] for span in recording_sink.spans] == ["first", "second", "third"] + + +def test_decorator_continuation_does_not_redeclare_or_replay( + otlp_logger: SplunkAOLogger, recording_sink: RecordingSink +) -> None: + # Given: independent decorator calls within one upstream request + upstream = trace.SpanContext(0x2468, 0x1357, True, trace.TraceFlags(1), trace.TraceState()) + token = context.attach(trace.set_span_in_context(trace.NonRecordingSpan(upstream))) + + @log(span_type="workflow", params={"metadata": lambda _: {"source": "decorator"}, "tags": lambda _: []}) + def operation() -> str: + return "answer" + + try: + with patch.object(splunk_ao_context, "get_logger_instance", return_value=otlp_logger): + # When: both top-level calls finish under the same OTel trace + operation() + operation() + # Then: both real operations exist, only the origin carries a declaration + first, continuation = recording_sink.spans + assert json.loads(first.attributes["splunk_ao.trace.metadata"]) == {"source": "decorator"} + assert first.attributes["splunk_ao.trace.tags"] == () + assert trace_attributes(continuation) == {} + assert first.parent == continuation.parent == upstream + finally: + context.detach(token) + + +def test_internal_envelope_preserves_timing_without_declaration( + otlp_logger: SplunkAOLogger, recording_sink: RecordingSink +) -> None: + # Given/When: integration bookkeeping supplies an internal timestamp + created_at = datetime(2000, 1, 1, tzinfo=UTC) + envelope = otlp_logger._start_trace_envelope(input="automatic", created_at=created_at, duration_ns=42) + otlp_logger.add_llm_span(input="actual", output="answer", model="model") + # Then: private timing remains supported without originating trace fields + assert envelope.created_at == created_at + assert envelope.metrics.duration_ns == 42 + assert trace_attributes(recording_sink.spans[0]) == {} + + +def test_hook_payload_excludes_private_declaration_and_preserves_dispatch() -> None: + # Given: legacy hook egress, with the same supplied manual fields + hook = Mock() + logger = SplunkAOLogger(project_id="project-id", agent_stream_id="log-stream-id", ingestion_hook=hook) + try: + envelope = logger.start_trace(input="question", name="support", metadata={"count": 3}) + logger.add_llm_span(input="prompt", output="answer", model="model") + logger.conclude(output="done") + # When: completed legacy payloads are dispatched + logger.flush() + # Then: normal envelope fields remain, with no wire snapshot leaking into models + hook.assert_called_once() + assert envelope.input == "question" + assert envelope.user_metadata == {"count": "3"} + assert "_trace_attributes" not in envelope.model_dump() + assert "splunk_ao.trace." not in json.dumps(envelope.model_dump(mode="json")) + finally: + logger.terminate() + + def test_complete_leaf_is_enqueued_before_flush(otlp_logger: SplunkAOLogger, recording_sink: RecordingSink) -> None: root = otlp_logger.start_trace(input="question") envelope_context = otlp_logger._otel_ids[root.id].span_context From 930bd5e36a7c607f314dad436a023ca418d8f44e Mon Sep 17 00:00:00 2001 From: pradystar Date: Fri, 6 Nov 2026 08:40:39 -0800 Subject: [PATCH 2/2] fix tests --- tests/test_agent_control_bridge.py | 5 ++-- tests/test_logger_batch.py | 43 +++++++++++++++--------------- 2 files changed, 25 insertions(+), 23 deletions(-) diff --git a/tests/test_agent_control_bridge.py b/tests/test_agent_control_bridge.py index bed25b11..3cfcc332 100644 --- a/tests/test_agent_control_bridge.py +++ b/tests/test_agent_control_bridge.py @@ -223,13 +223,14 @@ def external_provider(): fake_agent_control_modules["trace_context"].set_trace_context_provider(external_provider) + # Internal envelopes exercise bridge ownership without redeclaring the active OTel trace. logger_a = SplunkAOLogger(project="project_a", agent_stream="stream_a") - logger_a.start_trace(input="trace a") + logger_a._start_trace_envelope(input="trace a") logger_a.add_workflow_span(input="workflow a", name="workflow_a") bridge_a = setup_agent_control_bridge(logger_a) logger_b = SplunkAOLogger(project="project_b", agent_stream="stream_b") - logger_b.start_trace(input="trace b") + logger_b._start_trace_envelope(input="trace b") workflow_b = logger_b.add_workflow_span(input="workflow b", name="workflow_b") bridge_b = setup_agent_control_bridge(logger_b) diff --git a/tests/test_logger_batch.py b/tests/test_logger_batch.py index 6754af53..17c5f960 100644 --- a/tests/test_logger_batch.py +++ b/tests/test_logger_batch.py @@ -151,7 +151,7 @@ def test_single_span_trace_to_galileo( setup_mock_projects_client(mock_projects_client) setup_mock_logstreams_client(mock_logstreams_client) - created_at = datetime.datetime.now() + created_at = datetime.datetime(2000, 1, 1, tzinfo=datetime.UTC) metadata = {"key": "value"} logger = _legacy_logger(project="my_project", agent_stream="my_log_stream") logger.start_trace( @@ -179,11 +179,12 @@ def test_single_span_trace_to_galileo( assert trace.input == "input" assert trace.output == "output" assert trace.name == "test-trace" - assert trace.created_at == created_at + assert trace.created_at != created_at + assert span.created_at == created_at assert trace.user_metadata == metadata assert trace.status_code == 200 assert trace.spans == [span] - assert trace.metrics == Metrics(duration_ns=1000000) + assert trace.metrics == Metrics() assert logger.traces == [] assert logger._parent_stack == deque() @@ -326,7 +327,7 @@ def test_single_span_trace_to_galileo_experiment_id( setup_mock_projects_client(mock_projects_client) setup_mock_experiments_client(mock_experiments_client) - created_at = datetime.datetime.now() + created_at = datetime.datetime(2000, 1, 1, tzinfo=datetime.UTC) metadata = {"key": "value"} logger = _legacy_logger(project="my_project", experiment_id="6c4e3f7e-4a9a-4e7e-8c1f-3a9a3a9a3a9a") logger.start_trace( @@ -342,11 +343,11 @@ def test_single_span_trace_to_galileo_experiment_id( assert trace.input == "input" assert trace.output == "output" assert trace.name == "test-trace" - assert trace.created_at == created_at + assert trace.created_at != created_at assert trace.user_metadata == metadata assert trace.status_code == 200 assert trace.spans == [] - assert trace.metrics == Metrics(duration_ns=1000000) + assert trace.metrics == Metrics() assert logger.traces == [] assert logger._parent_stack == deque() @@ -392,11 +393,7 @@ def test_nested_span_trace_to_galileo( mock_traces_client_instance.ingest_traces.assert_called_once() payload = mock_traces_client_instance.ingest_traces.call_args.args[0] - expected_payload = TracesIngestRequest( - agent_stream_id=None, - experiment_id=None, - traces=[trace], - ) + expected_payload = TracesIngestRequest(agent_stream_id=None, experiment_id=None, traces=[trace]) assert payload == expected_payload assert logger.traces == [] assert logger._parent_stack == deque() @@ -442,7 +439,7 @@ def test_multi_span_trace_to_galileo( setup_mock_projects_client(mock_projects_client) setup_mock_logstreams_client(mock_logstreams_client) - created_at = datetime.datetime.now() + created_at = datetime.datetime(2000, 1, 1, tzinfo=datetime.UTC) metadata = {"key": "value"} logger = _legacy_logger(project="my_project", agent_stream="my_log_stream") logger.start_trace( @@ -490,7 +487,9 @@ def test_multi_span_trace_to_galileo( assert trace.input == "input" assert trace.output == "response2" assert trace.name == "test-trace" - assert trace.created_at == created_at + assert trace.created_at != created_at + assert workflow_span.created_at == created_at + assert second_span.created_at == created_at assert trace.user_metadata == metadata assert trace.status_code == 200 assert trace.spans == [workflow_span, second_span] @@ -510,7 +509,7 @@ async def test_single_span_trace_to_galileo_with_async( setup_mock_projects_client(mock_projects_client) setup_mock_logstreams_client(mock_logstreams_client) - created_at = datetime.datetime.now() + created_at = datetime.datetime(2000, 1, 1, tzinfo=datetime.UTC) metadata = {"key": "value"} def local_scorer(step: Trace | Span) -> int: @@ -548,11 +547,12 @@ def local_scorer(step: Trace | Span) -> int: assert trace.input == "input" assert trace.output == "output" assert trace.name == "test-trace" - assert trace.created_at == created_at + assert trace.created_at != created_at + assert span.created_at == created_at assert trace.user_metadata == metadata assert trace.status_code == 200 assert trace.spans == [span] - assert trace.metrics == Metrics(duration_ns=1000000) + assert trace.metrics == Metrics() assert logger.traces == [] assert logger._parent_stack == deque() @@ -2082,13 +2082,13 @@ def test_loggers_have_isolated_state( assert logger_a.has_active_trace() is False assert logger_b.has_active_trace() is False - # Both loggers build concurrent hierarchies - logger_a.start_trace(input="Trace A", name="trace_a") + # Internal envelopes test bookkeeping isolation without redeclaring the active OTel trace. + logger_a._start_trace_envelope(input="Trace A", name="trace_a") logger_a.add_workflow_span(input="Workflow A", name="workflow_a") assert logger_a.has_active_trace() is True assert logger_b.has_active_trace() is False - logger_b.start_trace(input="Trace B", name="trace_b") + logger_b._start_trace_envelope(input="Trace B", name="trace_b") logger_b.add_workflow_span(input="Workflow B", name="workflow_b") # Each logger sees its own current parent @@ -2134,8 +2134,9 @@ def test_reset_only_affects_own_logger( logger_a = _legacy_logger(project="project_a", agent_stream="stream_a") logger_b = _legacy_logger(project="project_b", agent_stream="stream_b") - logger_a.start_trace(input="Trace A", name="trace_a") - logger_b.start_trace(input="Trace B", name="trace_b") + # Internal envelopes test bookkeeping isolation without redeclaring the active OTel trace. + logger_a._start_trace_envelope(input="Trace A", name="trace_a") + logger_b._start_trace_envelope(input="Trace B", name="trace_b") logger_b.add_workflow_span(input="Workflow B", name="workflow_b") # Reset logger_a's parent tracking - logger_b unaffected