fix: honor gen_ai payload capture disable regardless of OTel env vars

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

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()

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)

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

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:

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

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()