diff --git a/AGENTS.md b/AGENTS.md index 0197b06..3358d3d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -88,14 +88,21 @@ Register via `pyproject.toml` entry points, then list the name in `HA_OBSERVABIL ## Span processor pipeline (default) ``` -SamplingSpanProcessor (DbControlSpanProcessor) - └─ ExcludeByAttributeSpanProcessor (drops traceableai.span_type=nospan) - └─ BatchSpanProcessor - └─ OTLPSpanExporter +GenAiPayloadScrubSpanProcessor (strips GenAI payload attrs when capture disabled) + └─ SamplingSpanProcessor (DbControlSpanProcessor) + └─ ExcludeByAttributeSpanProcessor (drops traceableai.span_type=nospan) + └─ BatchSpanProcessor + └─ OTLPSpanExporter ``` `DbControlSpanProcessor` — filters MySQL/PostgreSQL spans through control plugins. +`GenAiPayloadScrubSpanProcessor` — defense-in-depth: when `gen_ai.payload_capture_enabled` resolves to `false`, strips `gen_ai.input.messages`, `gen_ai.output.messages`, `gen_ai.system_instruction`, `gen_ai.prompt*`, `gen_ai.completion*`, `traceloop.entity.input`, and `traceloop.entity.output` from every span before export, regardless of which instrumentation set them. Cheap no-op when capture is enabled. + +### GenAI payload capture is a privacy control (disable always wins) + +`HARNESS_GEN_AI_PAYLOAD_CAPTURE_ENABLED=false` (or YAML `gen_ai.payload_capture_enabled: false`) must guarantee no prompt/response content is captured, even if the deployment environment already set OTel/Traceloop env vars (`OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT`, `OTEL_SEMCONV_STABILITY_OPT_IN`, `TRACELOOP_TRACE_CONTENT`) before the SDK initializes. Whenever the resolved config value is `false` (explicit or default), the SDK forces those env vars off and patches the cached OTel semconv-stability state, overwriting whatever was already there. When capture is enabled, the SDK only supplies defaults: an env var the user already set is left untouched. The `GenAiPayloadScrubSpanProcessor` above is the final backstop for instrumentation paths that don't consult config directly. + ## Build / vendor ```bash diff --git a/src/harness_sdk/gen_ai_payload_scrub_span_processor.py b/src/harness_sdk/gen_ai_payload_scrub_span_processor.py new file mode 100644 index 0000000..3086f69 --- /dev/null +++ b/src/harness_sdk/gen_ai_payload_scrub_span_processor.py @@ -0,0 +1,76 @@ +"""Defense-in-depth span processor that scrubs GenAI payload attributes. + +Instrumentation wrappers in this SDK gate payload-bearing attributes on +``gen_ai.payload_capture_enabled`` (see ``instrumentation/genai_env.py``, +``instrumentation/litellm/``, ``instrumentation/mcp/``), but third-party OTel +contrib instrumentations we do not wrap may not consult Harness config at all. +This processor is the last line of defense before export: when capture is +disabled it strips any payload-bearing GenAI/Traceloop attribute from every +span, regardless of which instrumentation set it. +""" +from opentelemetry.sdk.trace import SpanProcessor + +from harness_sdk.config.config import Config +from harness_sdk.custom_logger import get_custom_logger + +logger = get_custom_logger(__name__) + +_SCRUBBED_ATTRIBUTES = frozenset({ + "gen_ai.input.messages", + "gen_ai.output.messages", + "gen_ai.system_instruction", + "traceloop.entity.input", + "traceloop.entity.output", +}) +_SCRUBBED_PREFIXES = ("gen_ai.prompt", "gen_ai.completion") + + +def _is_payload_attribute(key: str) -> bool: + if key in _SCRUBBED_ATTRIBUTES: + return True + return any(key.startswith(prefix) for prefix in _SCRUBBED_PREFIXES) + + +class GenAiPayloadScrubSpanProcessor(SpanProcessor): + """Strips GenAI payload attributes from spans when capture is disabled. + + Cheap no-op passthrough when capture is enabled. Mutates the ended span's + underlying attribute store directly: ``ReadableSpan.attributes`` is a + read-only view (``MappingProxyType``) over the same dict the concrete SDK + ``Span`` still owns at ``on_end`` time, so deleting keys from it here is + reflected in whatever export path runs downstream. + """ + + def __init__(self, processor): + self._processor = processor + + def on_start(self, span, parent_context=None): + self._processor.on_start(span, parent_context) + + def on_end(self, span): + if Config().config.gen_ai.payload_capture_enabled.value: + self._processor.on_end(span) + return + self._scrub(span) + self._processor.on_end(span) + + @staticmethod + def _scrub(span) -> None: + attributes = getattr(span, "_attributes", None) + if not attributes: + return + scrubbed_keys = [key for key in attributes if _is_payload_attribute(key)] + for key in scrubbed_keys: + del attributes[key] + if scrubbed_keys: + logger.debug( + "GenAI: scrubbed payload attributes from span %s: %s", + span.name, + scrubbed_keys, + ) + + def force_flush(self, timeout_millis=30000): + return self._processor.force_flush(timeout_millis) + + def shutdown(self): + return self._processor.shutdown() diff --git a/src/harness_sdk/instrumentation/genai_env.py b/src/harness_sdk/instrumentation/genai_env.py index 6913d3e..002846c 100644 --- a/src/harness_sdk/instrumentation/genai_env.py +++ b/src/harness_sdk/instrumentation/genai_env.py @@ -12,22 +12,39 @@ _OTEL_GENAI_CAPTURE_VAR = "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT" _OTEL_SEMCONV_STABILITY_VAR = "OTEL_SEMCONV_STABILITY_OPT_IN" _GENAI_EXPERIMENTAL_VALUE = "gen_ai_latest_experimental" +_NO_CONTENT_VALUE = "NO_CONTENT" _applied: bool = False def maybe_set_genai_payload_capture_env_vars() -> None: - """Set OTEL payload-capture env vars from Traceable config if not already set by the user. + """Sync OTEL payload-capture env vars with the resolved Traceable config. Must be called before any GenAI instrumentation wrapper evaluates should_capture_content_on_spans_in_experimental_mode(), because the OTel semconv stability class caches its mode on first access. We also patch the cache directly to handle the case where OTel initialised before this runs. + + Precedence differs by direction, because payload capture is a privacy + control and "false" must always mean false: + - Disabled (``payload_capture_enabled`` resolves to False, whether via + explicit config or the untouched default): force capture off, + overwriting any pre-existing OTEL_* env vars and the semconv stability + cache. A deployment that pre-sets these vars (e.g. via a shared base + image or another OTel auto-instrumentation layer) must not be able to + resurrect capture. + - Enabled: only supply defaults. If the user already set either OTEL_* + var, leave both alone. """ global _applied # pylint: disable=global-statement if _applied: return + if not Config().config.gen_ai.payload_capture_enabled.value: + _force_disable_payload_capture() + _applied = True + return + capture_var_set = _OTEL_GENAI_CAPTURE_VAR in os.environ semconv_var_set = _OTEL_SEMCONV_STABILITY_VAR in os.environ if capture_var_set or semconv_var_set: @@ -37,10 +54,6 @@ def maybe_set_genai_payload_capture_env_vars() -> None: _applied = True return - if not Config().config.gen_ai.payload_capture_enabled.value: - _applied = True - return - os.environ[_OTEL_SEMCONV_STABILITY_VAR] = _GENAI_EXPERIMENTAL_VALUE os.environ[_OTEL_GENAI_CAPTURE_VAR] = "SPAN_ONLY" logger.debug( @@ -69,3 +82,37 @@ def maybe_set_genai_payload_capture_env_vars() -> None: logger.debug("GenAI: could not patch OTel semconv stability cache: %s", err) _applied = True + + +def _force_disable_payload_capture() -> None: + """Force GenAI content capture off, overwriting any pre-existing OTel env vars. + + Sets the capture-mode env var to NO_CONTENT (the value every OTel GenAI + instrumentation treats as "do not capture", regardless of which semconv + stability mode it ends up in) and patches the cached semconv stability + mode for the GEN_AI signal to DEFAULT (non-experimental), so instrumentation + that only checks "is experimental mode" also sees capture as off. Both are + forced unconditionally: this is the disable direction of a privacy control, + so config wins over whatever the deployment environment set. + """ + os.environ[_OTEL_GENAI_CAPTURE_VAR] = _NO_CONTENT_VALUE + logger.debug( + "GenAI: payload_capture_enabled=False; forcing %s=%s regardless of pre-existing env vars.", + _OTEL_GENAI_CAPTURE_VAR, + _NO_CONTENT_VALUE, + ) + + try: + from opentelemetry.instrumentation._semconv import ( # pylint: disable=import-outside-toplevel + _OpenTelemetrySemanticConventionStability, + _OpenTelemetryStabilitySignalType, + _StabilityMode, + ) + with _OpenTelemetrySemanticConventionStability._lock: # pylint: disable=protected-access + _OpenTelemetrySemanticConventionStability._OTEL_SEMCONV_STABILITY_SIGNAL_MAPPING[ # pylint: disable=protected-access + _OpenTelemetryStabilitySignalType.GEN_AI + ] = _StabilityMode.DEFAULT + _OpenTelemetrySemanticConventionStability._initialized = True # pylint: disable=protected-access + logger.debug("GenAI: patched OTel semconv stability cache to DEFAULT (non-experimental) for GEN_AI.") + except Exception as err: # pylint: disable=broad-except + logger.debug("GenAI: could not patch OTel semconv stability cache: %s", err) diff --git a/src/harness_sdk/instrumentation/litellm/__init__.py b/src/harness_sdk/instrumentation/litellm/__init__.py index 13e24fb..7455b16 100644 --- a/src/harness_sdk/instrumentation/litellm/__init__.py +++ b/src/harness_sdk/instrumentation/litellm/__init__.py @@ -224,6 +224,10 @@ def _set_pre_call_request_attributes( if pre_call.payload is None: return + # Harness config is authoritative in the disable direction: even if LiteLLM's + # own message_logging flag is on, a config-disabled capture must win. + if not Config().config.gen_ai.payload_capture_enabled.value: + return try: import litellm # pylint: disable=import-outside-toplevel from litellm.litellm_core_utils.safe_json_dumps import ( # pylint: disable=import-outside-toplevel diff --git a/src/harness_sdk/instrumentation/mcp/gen_ai_mirror.py b/src/harness_sdk/instrumentation/mcp/gen_ai_mirror.py index fc8b6b9..7447e1e 100644 --- a/src/harness_sdk/instrumentation/mcp/gen_ai_mirror.py +++ b/src/harness_sdk/instrumentation/mcp/gen_ai_mirror.py @@ -25,13 +25,19 @@ def apply_gen_ai_env_for_mcp() -> None: The MCP contrib package gates body capture on TRACELOOP_TRACE_CONTENT; we mirror TA_GEN_AI_* via Config (see environment.default / TA_GEN_AI_*). + + Disable is forced unconditionally, overwriting any pre-existing + TRACELOOP_TRACE_CONTENT (this is a privacy control: config-disabled must + always win over a deployment-provided env var). Enable only supplies a + default and leaves an existing env var untouched. """ gen = Config().config.gen_ai + if not gen.payload_capture_enabled.value: + os.environ["TRACELOOP_TRACE_CONTENT"] = "false" + return if "TRACELOOP_TRACE_CONTENT" in os.environ: return - os.environ["TRACELOOP_TRACE_CONTENT"] = ( - "true" if gen.payload_capture_enabled.value else "false" - ) + os.environ["TRACELOOP_TRACE_CONTENT"] = "true" def _mirror_span_kind(span, key: str, value: Any, tool_kind: str) -> None: diff --git a/src/harness_sdk/plugins/builtin/pipeline.py b/src/harness_sdk/plugins/builtin/pipeline.py index 46a3f45..8153b8c 100644 --- a/src/harness_sdk/plugins/builtin/pipeline.py +++ b/src/harness_sdk/plugins/builtin/pipeline.py @@ -9,6 +9,7 @@ from harness_sdk.custom_logger import get_custom_logger from harness_sdk.excluded_by_attribute_span_processor import ExcludeByAttributeSpanProcessor from harness_sdk.db_control_span_processor import DbControlSpanProcessor +from harness_sdk.gen_ai_payload_scrub_span_processor import GenAiPayloadScrubSpanProcessor logger = get_custom_logger(__name__) @@ -46,7 +47,10 @@ def create_span_processors(self, config: Any) -> List[SpanProcessor]: excluded_value="nospan", ) db_control_processor = DbControlSpanProcessor(filter_processor) - return [db_control_processor] + # Outermost: scrub GenAI payload attributes before any other on_end + # logic (control evaluation, filtering, batching) sees the span. + scrub_processor = GenAiPayloadScrubSpanProcessor(db_control_processor) + return [scrub_processor] def shutdown(self) -> None: pass diff --git a/test/gen_ai_payload_scrub_span_processor_test.py b/test/gen_ai_payload_scrub_span_processor_test.py new file mode 100644 index 0000000..7e4af7e --- /dev/null +++ b/test/gen_ai_payload_scrub_span_processor_test.py @@ -0,0 +1,115 @@ +import unittest +from unittest.mock import MagicMock, patch + +from opentelemetry.sdk.trace import TracerProvider + +from harness_sdk.gen_ai_payload_scrub_span_processor import GenAiPayloadScrubSpanProcessor + + +class TestGenAiPayloadScrubSpanProcessor(unittest.TestCase): + def setUp(self): + self.mock_processor = MagicMock() + self.processor = GenAiPayloadScrubSpanProcessor(processor=self.mock_processor) + self.tracer_provider = TracerProvider() + self.tracer = self.tracer_provider.get_tracer(__name__) + + def create_test_span(self, attributes=None): + """Helper method to create a test span with the given attributes.""" + with self.tracer.start_as_current_span("test-span") as span: + if attributes: + for key, value in attributes.items(): + span.set_attribute(key, value) + return span + + def _mock_config(self, payload_capture_enabled: bool): + gen = MagicMock() + gen.payload_capture_enabled.value = payload_capture_enabled + cfg = MagicMock() + cfg.gen_ai = gen + root = MagicMock() + root.config = cfg + return root + + def test_on_start_delegates_to_processor(self): + span = self.create_test_span() + parent_context = MagicMock() + + self.processor.on_start(span, parent_context) + + self.mock_processor.on_start.assert_called_once_with(span, parent_context) + + def test_capture_enabled_is_noop_passthrough(self): + span = self.create_test_span({ + "gen_ai.input.messages": "[secret prompt]", + "keep.me": "value", + }) + + with patch( + "harness_sdk.gen_ai_payload_scrub_span_processor.Config", + return_value=self._mock_config(payload_capture_enabled=True), + ): + self.processor.on_end(span) + + self.mock_processor.on_end.assert_called_once_with(span) + assert span.attributes.get("gen_ai.input.messages") == "[secret prompt]" + + def test_capture_disabled_strips_known_payload_attributes(self): + span = self.create_test_span({ + "gen_ai.input.messages": "[secret prompt]", + "gen_ai.output.messages": "[secret response]", + "gen_ai.system_instruction": "system prompt", + "gen_ai.prompt.0.content": "hi", + "gen_ai.completion.0.content": "hello", + "traceloop.entity.input": "input payload", + "traceloop.entity.output": "output payload", + "gen_ai.request.model": "gpt-4o-mini", + "gen_ai.usage.input_tokens": 3, + }) + + with patch( + "harness_sdk.gen_ai_payload_scrub_span_processor.Config", + return_value=self._mock_config(payload_capture_enabled=False), + ): + self.processor.on_end(span) + + attrs = span.attributes + for key in ( + "gen_ai.input.messages", + "gen_ai.output.messages", + "gen_ai.system_instruction", + "gen_ai.prompt.0.content", + "gen_ai.completion.0.content", + "traceloop.entity.input", + "traceloop.entity.output", + ): + assert key not in attrs, f"{key} should have been scrubbed" + + # Non-payload metadata attributes must survive the scrub. + assert attrs.get("gen_ai.request.model") == "gpt-4o-mini" + assert attrs.get("gen_ai.usage.input_tokens") == 3 + + self.mock_processor.on_end.assert_called_once_with(span) + + def test_capture_disabled_on_span_without_payload_attributes_is_harmless(self): + span = self.create_test_span({"gen_ai.request.model": "gpt-4o-mini"}) + + with patch( + "harness_sdk.gen_ai_payload_scrub_span_processor.Config", + return_value=self._mock_config(payload_capture_enabled=False), + ): + self.processor.on_end(span) + + assert span.attributes.get("gen_ai.request.model") == "gpt-4o-mini" + self.mock_processor.on_end.assert_called_once_with(span) + + def test_force_flush_delegates_to_processor(self): + self.processor.force_flush(5000) + self.mock_processor.force_flush.assert_called_once_with(5000) + + def test_shutdown_delegates_to_processor(self): + self.processor.shutdown() + self.mock_processor.shutdown.assert_called_once() + + +if __name__ == "__main__": + unittest.main() diff --git a/test/instrumentation/litellm/litellm_instrumentation_test.py b/test/instrumentation/litellm/litellm_instrumentation_test.py index 800d19a..0564b2f 100644 --- a/test/instrumentation/litellm/litellm_instrumentation_test.py +++ b/test/instrumentation/litellm/litellm_instrumentation_test.py @@ -96,6 +96,43 @@ def test_litellm_completion_span_has_gen_ai_attributes(agent, exporter, litellm_ assert attrs.get("gen_ai.usage.reasoning.output_tokens") == 1 +def test_litellm_input_messages_captured_when_enabled(agent, exporter, litellm_instrumentor): # pylint: disable=unused-argument + with patch("litellm.main.completion", new=_fake_model_response): + litellm_instrumentor.instrument() + litellm.completion( + model="gpt-4o-mini", + messages=[{"role": "user", "content": "hi"}], + ) + + spans = exporter.get_finished_spans() + exporter.clear() + attrs = _request_span(spans).attributes + assert "gen_ai.input.messages" in attrs + assert "hi" in attrs.get("gen_ai.input.messages") + + +def test_litellm_input_messages_omitted_when_capture_disabled(agent, exporter, litellm_instrumentor, monkeypatch): # pylint: disable=unused-argument + monkeypatch.setenv("HA_GEN_AI_PAYLOAD_CAPTURE_ENABLED", "false") + from harness_sdk.config.config import Config # pylint: disable=import-outside-toplevel + + Config._instance = None + + with patch("litellm.main.completion", new=_fake_model_response): + litellm_instrumentor.instrument() + litellm.completion( + model="gpt-4o-mini", + messages=[{"role": "user", "content": "hi"}], + ) + + spans = exporter.get_finished_spans() + exporter.clear() + attrs = _request_span(spans).attributes + assert "gen_ai.input.messages" not in attrs + # Non-payload request attributes are still recorded. + assert attrs.get("gen_ai.request.model") == "gpt-4o-mini" + assert attrs.get("gen_ai.operation.name") == "chat" + + def test_litellm_evaluate_blocks_before_wrapped(agent, exporter, litellm_instrumentor): # pylint: disable=unused-argument calls = {"n": 0} diff --git a/test/instrumentation/mcp/test_gen_ai_mirror.py b/test/instrumentation/mcp/test_gen_ai_mirror.py index 08b43bd..fb08e47 100644 --- a/test/instrumentation/mcp/test_gen_ai_mirror.py +++ b/test/instrumentation/mcp/test_gen_ai_mirror.py @@ -89,6 +89,15 @@ def test_apply_gen_ai_env_respects_existing_env(mock_gen_ai_config): assert os.environ.get("TRACELOOP_TRACE_CONTENT") == "true" +def test_apply_gen_ai_env_disable_overwrites_preexisting_true(mock_gen_ai_config): + """Disable is a privacy control: it must overwrite an env var the deployment + already set to true, unlike the enable path which only supplies a default.""" + mock_gen_ai_config.payload_capture_enabled.value = False + with patch.dict(os.environ, {"TRACELOOP_TRACE_CONTENT": "true"}): + apply_gen_ai_env_for_mcp() + assert os.environ.get("TRACELOOP_TRACE_CONTENT") == "false" + + def test_mirror_entity_name_skipped_when_not_tool_kind(mock_gen_ai_config): span = MagicMock() mirror_traceloop_to_gen_ai( diff --git a/test/instrumentation/test_genai_env.py b/test/instrumentation/test_genai_env.py new file mode 100644 index 0000000..94de886 --- /dev/null +++ b/test/instrumentation/test_genai_env.py @@ -0,0 +1,145 @@ +"""Tests for GenAI OTel env var forcing (harness_sdk.instrumentation.genai_env). + +Focus: config-disabled payload capture must force OTel content-capture off, +overwriting any pre-existing OTEL_* env vars and the OTel semconv stability +cache, while config-enabled must only supply defaults and never clobber env +vars the user already set. +""" +import os +from unittest.mock import MagicMock, patch + +import pytest + +from harness_sdk.instrumentation import genai_env as genai_env_mod +from harness_sdk.instrumentation.genai_env import ( + _OTEL_GENAI_CAPTURE_VAR, + _OTEL_SEMCONV_STABILITY_VAR, + maybe_set_genai_payload_capture_env_vars, +) +from opentelemetry.instrumentation._semconv import ( + _OpenTelemetrySemanticConventionStability, + _OpenTelemetryStabilitySignalType, + _StabilityMode, +) + + +@pytest.fixture(autouse=True) +def _isolate_genai_env_state(): + """Reset the module-level `_applied` flag and the OTel semconv cache. + + Both are process-global mutable state that would otherwise leak between + tests (and between test modules, since other instrumentation suites also + call `maybe_set_genai_payload_capture_env_vars`). + """ + genai_env_mod._applied = False + saved_capture_var = os.environ.pop(_OTEL_GENAI_CAPTURE_VAR, None) + saved_semconv_var = os.environ.pop(_OTEL_SEMCONV_STABILITY_VAR, None) + saved_mapping = dict( + _OpenTelemetrySemanticConventionStability._OTEL_SEMCONV_STABILITY_SIGNAL_MAPPING # pylint: disable=protected-access + ) + saved_initialized = _OpenTelemetrySemanticConventionStability._initialized # pylint: disable=protected-access + + yield + + genai_env_mod._applied = False + for key, value in ( + (_OTEL_GENAI_CAPTURE_VAR, saved_capture_var), + (_OTEL_SEMCONV_STABILITY_VAR, saved_semconv_var), + ): + if value is None: + os.environ.pop(key, None) + else: + os.environ[key] = value + _OpenTelemetrySemanticConventionStability._OTEL_SEMCONV_STABILITY_SIGNAL_MAPPING = saved_mapping # pylint: disable=protected-access + _OpenTelemetrySemanticConventionStability._initialized = saved_initialized # pylint: disable=protected-access + + +def _mock_config(payload_capture_enabled: bool): + gen = MagicMock() + gen.payload_capture_enabled.value = payload_capture_enabled + cfg = MagicMock() + cfg.gen_ai = gen + root = MagicMock() + root.config = cfg + return root + + +def _gen_ai_mapping(): + return _OpenTelemetrySemanticConventionStability._OTEL_SEMCONV_STABILITY_SIGNAL_MAPPING.get( # pylint: disable=protected-access + _OpenTelemetryStabilitySignalType.GEN_AI + ) + + +def test_disable_forces_no_content_overwriting_preset_env_vars(): + os.environ[_OTEL_GENAI_CAPTURE_VAR] = "SPAN_ONLY" + os.environ[_OTEL_SEMCONV_STABILITY_VAR] = "gen_ai_latest_experimental" + + with patch( + "harness_sdk.instrumentation.genai_env.Config", + return_value=_mock_config(payload_capture_enabled=False), + ): + maybe_set_genai_payload_capture_env_vars() + + assert os.environ.get(_OTEL_GENAI_CAPTURE_VAR) == "NO_CONTENT" + + +def test_disable_patches_semconv_cache_to_default(): + _OpenTelemetrySemanticConventionStability._initialized = True # pylint: disable=protected-access + _OpenTelemetrySemanticConventionStability._OTEL_SEMCONV_STABILITY_SIGNAL_MAPPING[ # pylint: disable=protected-access + _OpenTelemetryStabilitySignalType.GEN_AI + ] = _StabilityMode.GEN_AI_LATEST_EXPERIMENTAL + + with patch( + "harness_sdk.instrumentation.genai_env.Config", + return_value=_mock_config(payload_capture_enabled=False), + ): + maybe_set_genai_payload_capture_env_vars() + + assert _gen_ai_mapping() == _StabilityMode.DEFAULT + assert _OpenTelemetrySemanticConventionStability._initialized is True # pylint: disable=protected-access + + +def test_disable_with_no_preexisting_env_vars_still_forces_no_content(): + with patch( + "harness_sdk.instrumentation.genai_env.Config", + return_value=_mock_config(payload_capture_enabled=False), + ): + maybe_set_genai_payload_capture_env_vars() + + assert os.environ.get(_OTEL_GENAI_CAPTURE_VAR) == "NO_CONTENT" + assert _gen_ai_mapping() == _StabilityMode.DEFAULT + + +def test_enable_respects_preexisting_env_vars(): + os.environ[_OTEL_GENAI_CAPTURE_VAR] = "EVENT_ONLY" + os.environ[_OTEL_SEMCONV_STABILITY_VAR] = "some_user_value" + + with patch( + "harness_sdk.instrumentation.genai_env.Config", + return_value=_mock_config(payload_capture_enabled=True), + ): + maybe_set_genai_payload_capture_env_vars() + + assert os.environ.get(_OTEL_GENAI_CAPTURE_VAR) == "EVENT_ONLY" + assert os.environ.get(_OTEL_SEMCONV_STABILITY_VAR) == "some_user_value" + + +def test_enable_sets_defaults_when_absent(): + with patch( + "harness_sdk.instrumentation.genai_env.Config", + return_value=_mock_config(payload_capture_enabled=True), + ): + maybe_set_genai_payload_capture_env_vars() + + assert os.environ.get(_OTEL_GENAI_CAPTURE_VAR) == "SPAN_ONLY" + assert os.environ.get(_OTEL_SEMCONV_STABILITY_VAR) == "gen_ai_latest_experimental" + assert _gen_ai_mapping() == _StabilityMode.GEN_AI_LATEST_EXPERIMENTAL + + +def test_applied_flag_short_circuits_subsequent_calls(): + mock_config_cls = MagicMock(return_value=_mock_config(payload_capture_enabled=False)) + with patch("harness_sdk.instrumentation.genai_env.Config", mock_config_cls): + maybe_set_genai_payload_capture_env_vars() + maybe_set_genai_payload_capture_env_vars() + + assert mock_config_cls.call_count == 1 diff --git a/test/plugins/builtin/test_pipeline.py b/test/plugins/builtin/test_pipeline.py new file mode 100644 index 0000000..f7674cc --- /dev/null +++ b/test/plugins/builtin/test_pipeline.py @@ -0,0 +1,27 @@ +"""Tests for the default observability pipeline wiring (builtin_pipeline plugin).""" +from harness_sdk.config.config import Config +from harness_sdk.db_control_span_processor import DbControlSpanProcessor +from harness_sdk.excluded_by_attribute_span_processor import ExcludeByAttributeSpanProcessor +from harness_sdk.gen_ai_payload_scrub_span_processor import GenAiPayloadScrubSpanProcessor +from harness_sdk.plugins.builtin.pipeline import BuiltinPipelinePlugin + + +def test_scrub_processor_wraps_chain_as_outermost_layer(monkeypatch): + # Force the real OTLP-export branch (skip the console-exporter early-return) + # so the full processor chain gets assembled. + monkeypatch.delenv("HA_ENABLE_CONSOLE_SPAN_EXPORTER", raising=False) + + config = Config() + plugin = BuiltinPipelinePlugin() + plugin.on_init(config) + + processors = plugin.create_span_processors(config) + + assert len(processors) == 1 + scrub_processor = processors[0] + assert isinstance(scrub_processor, GenAiPayloadScrubSpanProcessor) + # pylint: disable=protected-access + db_control_processor = scrub_processor._processor + assert isinstance(db_control_processor, DbControlSpanProcessor) + filter_processor = db_control_processor._processor + assert isinstance(filter_processor, ExcludeByAttributeSpanProcessor)