feat: flatten dict values passed to set_span_attribute

AGENTS.md

@@ -38,6 +38,8 @@ Key variables:
 | `HARNESS_CONFIG_FILE` | Path to YAML config file (overrides env) |
 | `HARNESS_GEN_AI_PAYLOAD_CAPTURE_ENABLED` | Capture LLM prompt/response payloads (default: off) |
 | `HARNESS_GEN_AI_PAYLOAD_EVALUATION_ENABLED` | Run control plugins on GenAI spans |
+| `HARNESS_SPAN_ATTRIBUTE_FLATTEN_ENABLED` | Flatten dict values from `set_span_attribute` into dot-notation keys (default: on) |
+| `HARNESS_SPAN_ATTRIBUTE_FLATTEN_RAW_JSON` | Also keep the original key as a JSON string when flattening (default: off) |
 
 ### Instrumentation opt-in (`HARNESS_` or `HA_` prefix; `HARNESS_` wins — no `AT_`/`TA_` aliases)
 

README.md

@@ -168,6 +168,36 @@ For duplicate keys, the last write wins; instrumentation can overwrite a custome
 if it writes the same key later. Because the span already exists, these attributes
 cannot influence head sampling.
 
+#### Dictionary values
+
+Dictionaries are also accepted and are flattened into dot-notation attributes so each
+leaf stays individually queryable in the backend:
+
+```python
+set_span_attribute("agent", {"action": "generate", "model": {"name": "gemini-2.0"}})
+# exported as: agent.action="generate", agent.model.name="gemini-2.0"
+```
+
+Nothing is serialized when you call the helper — the dict is flattened at span end, just
+before export. Flattening rules:
+
+| Case | Result |
+|---|---|
+| `str` / `bool` / `int` / `float` leaf | kept as the native OTel type |
+| `None` leaf | skipped |
+| Any other object | `str(value)` |
+| List of same-typed scalars | OTel array attribute at that key |
+| List of dicts or mixed types | JSON string at that key |
+| Nesting deeper than 3 levels | JSON string at the depth-3 key |
+| Flattened key already set on the span | skipped — the explicit value wins |
+
+At most 32 leaf attributes are emitted per dictionary; the rest are dropped with a debug
+log. The original key (`agent`) is not set unless
+`HARNESS_SPAN_ATTRIBUTE_FLATTEN_RAW_JSON=true`, which additionally stores the whole dict
+as JSON there. Flattening is **enabled by default**; set
+`HARNESS_SPAN_ATTRIBUTE_FLATTEN_ENABLED=false` to turn it off, in which case OTel rejects
+dictionary values as it did before.
+
 ## Plugins
 
 The SDK loads extensions via [setuptools entry points](https://setuptools.pypa.io/en/latest/userguide/entry_point.html). Each plugin has a **name** (the entry-point key). Names are listed in config or environment variables; only installed plugins are loaded, in the order you configure.

src/harness_sdk/agent_init.py

@@ -18,6 +18,8 @@
 from harness_sdk import constants
 from harness_sdk.config import config_pb2
 from harness_sdk.env import is_env_flag_enabled
+from harness_sdk.flatten_dict_registry import is_flatten_enabled
+from harness_sdk.flatten_dict_span_processor import FlattenDictSpanProcessor
 from harness_sdk.otlp_reporting import (
     compression_type_to_otlp_grpc,
     compression_type_to_otlp_http,
@@ -97,8 +99,10 @@ def register_processor(self, processor) -> None:
     def set_console_span_processor(self) -> None:
         console_span_exporter = ConsoleSpanExporter(
             service_name=self._config.config.service_name)
-        simple_export_span_processor = SimpleSpanProcessor(console_span_exporter)
-        trace.get_tracer_provider().add_span_processor(simple_export_span_processor)
+        processor = SimpleSpanProcessor(console_span_exporter)
+        if is_flatten_enabled():
+            processor = FlattenDictSpanProcessor(processor)
+        trace.get_tracer_provider().add_span_processor(processor)
 
     def init_exporter(self, trace_reporter_type):
         exporter_type = ''

src/harness_sdk/flatten_dict_registry.py

@@ -0,0 +1,98 @@
+"""Thread-safe hand-off of dict span attributes from the hot path to span end.
+
+``set_span_attribute("agent", {...})`` cannot go through
+``span.set_attribute``: OTel rejects mapping values outright. Serializing on
+the caller's thread would put JSON encoding on the application hot path, so
+dict values are parked here untouched and flattened later by
+``FlattenDictSpanProcessor.on_end``.
+
+Entries are keyed by span context rather than object identity because
+``Span.end()`` hands ``on_end`` a fresh ``ReadableSpan`` snapshot, not the
+recording ``Span`` the enrichment helper saw.
+"""
+import threading
+from collections import OrderedDict
+
+from harness_sdk.custom_logger import get_custom_logger
+from harness_sdk.env import get_env_value, is_env_flag_enabled
+
+logger = get_custom_logger(__name__)
+
+FLATTEN_ENABLED_ENV = "SPAN_ATTRIBUTE_FLATTEN_ENABLED"
+FLATTEN_RAW_JSON_ENV = "SPAN_ATTRIBUTE_FLATTEN_RAW_JSON"
+
+# Bound on spans holding pending dicts. A span that is never ended would
+# otherwise leak its entry forever; oldest entries are evicted instead.
+_MAX_TRACKED_SPANS = 2048
+
+
+def is_flatten_enabled():
+    """Dict flattening is on by default; only explicit ``false`` disables it."""
+    value = get_env_value(FLATTEN_ENABLED_ENV)
+    if value is None:
+        return True
+    return value.strip().lower() != "false"
+
+
+def is_raw_json_enabled():
+    """Opt in to additionally keeping the original key as a JSON string."""
+    return is_env_flag_enabled(FLATTEN_RAW_JSON_ENV)
+
+
+def _span_key(span):
+    get_context = getattr(span, "get_span_context", None)
+    if get_context is None:
+        return None
+    context = get_context()
+    if context is None or not context.trace_id:
+        return None
+    return (context.trace_id, context.span_id)
+
+
+class FlattenDictRegistry:
+    """Maps span identity to the dict attributes awaiting flattening."""
+
+    def __init__(self, max_tracked_spans=_MAX_TRACKED_SPANS):
+        self._lock = threading.Lock()
+        self._pending = OrderedDict()
+        self._max_tracked_spans = max_tracked_spans
+
+    def register(self, span, key, value):
+        """Park ``value`` under ``key`` for ``span``; last write wins."""
+        span_key = _span_key(span)
+        if span_key is None:
+            return
+        with self._lock:
+            attributes = self._pending.get(span_key)
+            if attributes is None:
+                attributes = OrderedDict()
+                self._pending[span_key] = attributes
+            attributes[key] = value
+            self._pending.move_to_end(span_key)
+            while len(self._pending) > self._max_tracked_spans:
+                evicted, _ = self._pending.popitem(last=False)
+                logger.debug(
+                    "Flatten: evicted pending dict attributes for span %s "
+                    "(registry limit %s reached)",
+                    evicted,
+                    self._max_tracked_spans,
+                )
+
+    def pop(self, span):
+        """Remove and return the pending dict attributes for ``span``."""
+        span_key = _span_key(span)
+        if span_key is None:
+            return {}
+        with self._lock:
+            return self._pending.pop(span_key, {})
+
+    def clear(self):
+        with self._lock:
+            self._pending.clear()
+
+
+_REGISTRY = FlattenDictRegistry()
+
+
+def get_registry():
+    return _REGISTRY

src/harness_sdk/flatten_dict_span_processor.py

@@ -0,0 +1,147 @@
+"""Span processor that expands dict span attributes into dot-notation keys.
+
+``set_span_attribute("agent", {"action": "generate"})`` parks the dict in
+``flatten_dict_registry`` instead of handing it to OTel, which would reject it.
+This processor drains the registry at ``on_end`` and writes
+``agent.action=generate`` so the backend gets individually queryable
+attributes instead of an opaque JSON blob.
+
+It must be the outermost processor: downstream scrubbing and exclusion logic
+matches on attribute keys, so the flattened keys have to exist before those
+run. Mutating the ended span works the same way ``GenAiPayloadScrubSpanProcessor``
+relies on: ``ReadableSpan.attributes`` is a read-only view over the attribute
+store the concrete SDK ``Span`` still owns at ``on_end`` time.
+"""
+import json
+from typing import Mapping
+
+from opentelemetry.sdk.trace import SpanProcessor
+
+from harness_sdk.custom_logger import get_custom_logger
+from harness_sdk.flatten_dict_registry import (
+    get_registry,
+    is_raw_json_enabled,
+)
+
+logger = get_custom_logger(__name__)
+
+MAX_DEPTH = 3
+MAX_LEAF_ATTRIBUTES = 32
+
+_SCALAR_TYPES = (bool, int, float, str)
+
+
+def _is_scalar(value):
+    return isinstance(value, _SCALAR_TYPES)
+
+
+def _scalar_kind(value):
+    # bool is a subclass of int, but OTel treats them as distinct array types.
+    if isinstance(value, bool):
+        return bool
+    if isinstance(value, int):
+        return int
+    if isinstance(value, float):
+        return float
+    return str
+
+
+def _to_json(value):
+    try:
+        return json.dumps(value, default=str)
+    except (TypeError, ValueError):
+        return str(value)
+
+
+def _sequence_leaf(value):
+    """Homogeneous scalar sequences stay arrays; anything else becomes JSON."""
+    items = tuple(value)
+    if not items:
+        return items
+    kinds = {_scalar_kind(item) for item in items if _is_scalar(item)}
+    if len(kinds) == 1 and all(_is_scalar(item) for item in items):
+        return items
+    return _to_json(value)
+
+
+def _leaf_value(value):
+    """Convert a non-mapping value to something OTel accepts, or None to skip."""
+    if value is None:
+        return None
+    if _is_scalar(value):
+        return value
+    if isinstance(value, (list, tuple, set, frozenset)):
+        return _sequence_leaf(value)
+    return str(value)
+
+
+def _collect(prefix, mapping, depth, flattened):
+    """Walk ``mapping`` into ``flattened``; returns False once the cap is hit."""
+    for key, value in mapping.items():
+        if len(flattened) >= MAX_LEAF_ATTRIBUTES:
+            return False
+        flat_key = f"{prefix}.{key}"
+        if isinstance(value, Mapping):
+            if depth < MAX_DEPTH:
+                if not _collect(flat_key, value, depth + 1, flattened):
+                    return False
+            else:
+                flattened[flat_key] = _to_json(value)
+            continue
+        leaf = _leaf_value(value)
+        if leaf is not None:
+            flattened[flat_key] = leaf
+    return True
+
+
+class FlattenDictSpanProcessor(SpanProcessor):
+    """Flattens registered dict attributes onto the span before export."""
+
+    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):
+        pending = get_registry().pop(span)
+        if pending:
+            try:
+                self._flatten(span, pending)
+            except Exception as err:  # pylint: disable=W0703
+                logger.debug(
+                    "Flatten: failed to flatten dict attributes on span %s: %s",
+                    getattr(span, "name", None),
+                    err,
+                )
+        self._processor.on_end(span)
+
+    @staticmethod
+    def _flatten(span, pending):
+        attributes = getattr(span, "_attributes", None)
+        if attributes is None:
+            return
+        raw_json = is_raw_json_enabled()
+        for root_key, value in pending.items():
+            flattened = {}
+            if not _collect(root_key, value, 1, flattened):
+                logger.debug(
+                    "Flatten: dict attribute %r on span %s exceeded %s leaf "
+                    "attributes; remaining entries dropped",
+                    root_key,
+                    getattr(span, "name", None),
+                    MAX_LEAF_ATTRIBUTES,
+                )
+            for flat_key, leaf in flattened.items():
+                # An explicitly set attribute always wins over a flattened one.
+                if flat_key in attributes:
+                    continue
+                attributes[flat_key] = leaf
+            if raw_json and root_key not in attributes:
+                attributes[root_key] = _to_json(value)
+
+    def force_flush(self, timeout_millis=30000):
+        return self._processor.force_flush(timeout_millis)
+
+    def shutdown(self):
+        return self._processor.shutdown()

src/harness_sdk/plugins/builtin/pipeline.py

@@ -9,6 +9,8 @@
 from harness_sdk.env import is_env_flag_enabled
 from harness_sdk.excluded_by_attribute_span_processor import ExcludeByAttributeSpanProcessor
 from harness_sdk.db_control_span_processor import DbControlSpanProcessor
+from harness_sdk.flatten_dict_registry import is_flatten_enabled
+from harness_sdk.flatten_dict_span_processor import FlattenDictSpanProcessor
 from harness_sdk.gen_ai_payload_scrub_span_processor import GenAiPayloadScrubSpanProcessor
 
 logger = get_custom_logger(__name__)
@@ -43,10 +45,14 @@ def create_span_processors(self, config: Any) -> List[SpanProcessor]:
             excluded_value="nospan",
         )
         db_control_processor = DbControlSpanProcessor(filter_processor)
-        # Outermost: scrub GenAI payload attributes before any other on_end
-        # logic (control evaluation, filtering, batching) sees the span.
+        # Scrub GenAI payload attributes before control evaluation, filtering
+        # and batching see the span.
         scrub_processor = GenAiPayloadScrubSpanProcessor(db_control_processor)
-        return [scrub_processor]
+        if not is_flatten_enabled():
+            return [scrub_processor]
+        # Outermost: dict attributes must be expanded into their flat keys
+        # before scrubbing and exclusion match on attribute keys.
+        return [FlattenDictSpanProcessor(scrub_processor)]
 
     def shutdown(self) -> None:
         pass

src/harness_sdk/span_enrichment.py

@@ -1,22 +1,35 @@
 """Public helpers for enriching the current OpenTelemetry span."""
 
-from typing import Mapping
+from typing import Any, Mapping, Union
 
 from opentelemetry import trace
 from opentelemetry.util.types import AttributeValue
 
+from harness_sdk.flatten_dict_registry import get_registry, is_flatten_enabled
 
-def set_span_attribute(key: str, value: AttributeValue) -> None:
+EnrichmentValue = Union[AttributeValue, Mapping[str, Any]]
+
+
+def set_span_attribute(key: str, value: EnrichmentValue) -> None:
     """Set one attribute on the current recording span."""
     span = trace.get_current_span()
     if span.is_recording():
-        span.set_attribute(key, value)
+        _set(span, key, value)
 
 
-def set_span_attributes(attributes: Mapping[str, AttributeValue]) -> None:
+def set_span_attributes(attributes: Mapping[str, EnrichmentValue]) -> None:
     """Set attributes on the current recording span."""
     span = trace.get_current_span()
     if not span.is_recording():
         return
     for key, value in attributes.items():
-        span.set_attribute(key, value)
+        _set(span, key, value)
+
+
+def _set(span, key: str, value: EnrichmentValue) -> None:
+    # Dict values are parked for FlattenDictSpanProcessor to expand into
+    # dot-notation keys at span end; nothing is serialized on this thread.
+    if isinstance(value, Mapping) and is_flatten_enabled():
+        get_registry().register(span, key, value)
+        return
+    span.set_attribute(key, value)