symfonic.services.observability.trace_policy

trace_policy

Tracing policy: metadata by default, content only by explicit consent.

TraceMode

Bases: StrEnum

How much diagnostic information may leave the process.

TracePolicy dataclass

TracePolicy(mode: TraceMode = TraceMode.METADATA, sample_rate: float = 1.0, failed_turns_always: bool = True, content_ttl_hours: int = 24, max_artifact_bytes: int = 262144)

One deployment's tracing, sampling and content-retention contract.

from_config classmethod

from_config(config: Any) -> TracePolicy

Build from modern trace_* fields, then legacy OTEL flags.

Source code in src/symfonic/services/observability/trace_policy.py

@classmethod
def from_config(cls, config: Any) -> TracePolicy:
    """Build from modern ``trace_*`` fields, then legacy OTEL flags."""
    raw_mode = getattr(config, "trace_mode", None)
    if raw_mode is None:
        raw_mode = getattr(config, "symfonic_trace_mode", None)
    if raw_mode is None:
        if not bool(getattr(config, "otel_enabled", False)):
            mode = TraceMode.OFF
        elif bool(getattr(config, "otel_capture_prompts", False)):
            mode = TraceMode.CONTENT
        else:
            mode = TraceMode.METADATA
    else:
        mode = TraceMode(str(raw_mode).lower())
    def configured(name: str, default: Any) -> Any:
        value = getattr(config, f"trace_{name}", None)
        if value is None:
            value = getattr(config, f"symfonic_trace_{name}", default)
        return value

    return cls(
        mode=mode,
        sample_rate=float(configured("sample_rate", 1.0)),
        failed_turns_always=bool(configured("failed_turns_always", True)),
        content_ttl_hours=int(configured("content_ttl_hours", 24)),
        max_artifact_bytes=int(configured("max_artifact_bytes", 262_144)),
    )

redact_trace_value

redact_trace_value(value: Any, *, max_string_bytes: int = 32768) -> Any

Recursively redact credentials and bound strings before export.

The result is JSON-shaped. Unknown objects are represented by their type, never by repr: a repr is allowed to contain precisely the secret the redactor is meant to keep out of the collector.

Source code in src/symfonic/services/observability/trace_policy.py

def redact_trace_value(value: Any, *, max_string_bytes: int = 32_768) -> Any:
    """Recursively redact credentials and bound strings before export.

    The result is JSON-shaped. Unknown objects are represented by their type,
    never by ``repr``: a repr is allowed to contain precisely the secret the
    redactor is meant to keep out of the collector.
    """
    if isinstance(value, Mapping):
        return {
            str(key): (
                "[REDACTED]"
                if _SECRET_KEY.search(str(key))
                else redact_trace_value(item, max_string_bytes=max_string_bytes)
            )
            for key, item in value.items()
        }
    if isinstance(value, (list, tuple, set, frozenset)):
        return [
            redact_trace_value(item, max_string_bytes=max_string_bytes)
            for item in value
        ]
    if isinstance(value, bytes):
        return f"[bytes:{len(value)}]"
    if isinstance(value, str):
        clean = _PRIVATE_KEY.sub("[REDACTED PRIVATE KEY]", value)
        clean = _BEARER.sub("Bearer [REDACTED]", clean)
        encoded = clean.encode("utf-8")
        if len(encoded) <= max_string_bytes:
            return clean
        suffix = "...[TRUNCATED]"
        room = max_string_bytes - len(suffix.encode("utf-8"))
        return encoded[:room].decode("utf-8", errors="ignore") + suffix
    if value is None or isinstance(value, (bool, int, float)):
        return value
    return f"[{type(value).__name__}]"

strip_trace_reasoning

strip_trace_reasoning(value: Any) -> Any

Recursively remove provider reasoning blocks before trace export.

Provider history can carry reasoning under nested content blocks or additional metadata. Redaction is not sufficient: raw chain-of-thought is data the tracing contract never collects, even in content mode.

Source code in src/symfonic/services/observability/trace_policy.py

def strip_trace_reasoning(value: Any) -> Any:
    """Recursively remove provider reasoning blocks before trace export.

    Provider history can carry reasoning under nested content blocks or
    additional metadata.  Redaction is not sufficient: raw chain-of-thought
    is data the tracing contract never collects, even in content mode.
    """

    def clean(item: Any) -> Any:
        if isinstance(item, Mapping):
            block_type = str(item.get("type", "")).casefold()
            if block_type in _REASONING_BLOCKS:
                return _DROP
            event_kind = str(item.get("kind", "")).casefold()
            result: dict[str, Any] = {}
            for key, nested in item.items():
                normalized = str(key).casefold()
                if normalized in _REASONING_KEYS:
                    continue
                if event_kind in _REASONING_BLOCKS and normalized in {
                    "content", "data", "payload", "text",
                }:
                    continue
                cleaned = clean(nested)
                if cleaned is not _DROP:
                    result[str(key)] = cleaned
            return result
        if isinstance(item, (list, tuple, set, frozenset)):
            return [cleaned for nested in item if (cleaned := clean(nested)) is not _DROP]
        return item

    cleaned = clean(value)
    return None if cleaned is _DROP else cleaned

trace_has_reasoning

trace_has_reasoning(value: Any) -> bool

Whether a trace-shaped value contains reasoning rather than metadata.

Source code in src/symfonic/services/observability/trace_policy.py

def trace_has_reasoning(value: Any) -> bool:
    """Whether a trace-shaped value contains reasoning rather than metadata."""
    if isinstance(value, str) and value.lstrip().startswith(("{", "[")):
        try:
            decoded = json.loads(value)
        except ValueError:
            decoded = None
        if isinstance(decoded, (Mapping, list)):
            return trace_has_reasoning(decoded)
    if isinstance(value, Mapping):
        block_type = str(value.get("type", "")).casefold()
        if block_type in _REASONING_BLOCKS:
            return True
        event_kind = str(value.get("kind", "")).casefold()
        for key, nested in value.items():
            normalized = str(key).casefold()
            if normalized in _REASONING_KEYS:
                return True
            if (
                event_kind in _REASONING_BLOCKS
                and normalized in {"content", "data", "payload", "text"}
                and nested
            ):
                return True
            if trace_has_reasoning(nested):
                return True
        return False
    if isinstance(value, (list, tuple, set, frozenset)):
        return any(trace_has_reasoning(item) for item in value)
    return False