symfonic.services.observability.capture

capture

Prompt-capture policy — the one place "may I record this?" is answered.

Before this module the answer was a bare otel_capture_prompts boolean read at each emission site. A boolean read in four places is four policies that happen to agree, and the day one site forgets to read it is the day prompts leak into a span exporter nobody audited.

Where the gate belongs. otel_capture_prompts has always governed one thing: whether prompt and completion text is stamped onto a span attribute and shipped to a collector. The shipped SpanRecorder still enforces it there. It has never governed what an in-process CallbackHandler is handed — emit_llm_end populates LLMEndEvent.output unconditionally — and in-process delivery to code the adopter registered themselves is not an egress boundary. Reading that flag before fan-out would therefore silently empty every adopter payload the moment the run source moved to this bridge: a handler persisting completions for quality review would keep running and start recording empty strings, with no exception and no log line.

So :meth:PromptCapturePolicy.from_config delivers payloads, and redaction before fan-out is a separate, explicit opt-in (FrameworkConfig.observability_redact_payloads) for adopters whose observers are themselves an egress boundary. Both answers remain one object rather than a boolean re-read per site — the split is which question it answers, not how many places answer it.

That opt-in is a declared field on FrameworkConfig, and it has to be: the model is extra="ignore", so an undeclared keyword is dropped at construction with nothing but a warning. A redaction control that is only ever read through getattr is a control that fails open — the adopter sets it, pydantic discards it, and every payload ships anyway. from_config still reads it defensively so a duck-typed config object works, but the supported surface is the field.

Being declared is necessary and not sufficient: a control nothing on the wiring path calls fails open just as quietly. compose_event_sink therefore takes the config itself and derives the policy from it, so the recipe adopters are given honours the flag without anyone remembering to construct a PromptCapturePolicy by hand. See :mod:.suite.

PromptCapturePolicy dataclass

PromptCapturePolicy(capture_prompts: bool = False, capture_outputs: bool = False, capture_tool_arguments: bool = False)

Three independent gates over the three payload classes.

They are independent because the risks are: an adopter may want the model's output for quality review while never recording the user's prompt, and a single flag cannot express that. from_config still maps the shipped single flag onto all three, so the current configuration surface is unchanged — the split is available, not imposed.

argument_names staticmethod

argument_names(arguments: Mapping[str, Any] | None) -> tuple[str, ...]

Sorted argument names, always available.

Names are structure; values are payload. Keeping the names when the values are denied is what lets an operator see that a tool was called with a customer_id without seeing which customer.

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

@staticmethod
def argument_names(arguments: Mapping[str, Any] | None) -> tuple[str, ...]:
    """Sorted argument *names*, always available.

    Names are structure; values are payload. Keeping the names when the
    values are denied is what lets an operator see that a tool was called
    with a ``customer_id`` without seeing which customer.
    """
    if not arguments:
        return ()
    return tuple(sorted(str(key) for key in arguments))

from_config classmethod

from_config(config: Any) -> PromptCapturePolicy

The policy for in-process fan-out, which is not an egress boundary.

Deliberately does not read otel_capture_prompts: that flag gates span attributes and is applied by the shipped SpanRecorder at the exporter, where the data actually leaves the process. Reusing it here would broaden its meaning from "do not put prompts in span attributes" to "do not give any consumer the text", silently emptying callback payloads that shipped code has always populated.

FrameworkConfig.observability_redact_payloads is the explicit opt-in for adopters who want the bridge itself to withhold. It is read through getattr so a duck-typed config still works, but it is a declared field precisely because a getattr-only control on an extra="ignore" model would be silently dropped and fail open.

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

@classmethod
def from_config(cls, config: Any) -> PromptCapturePolicy:
    """The policy for *in-process fan-out*, which is not an egress boundary.

    Deliberately does **not** read ``otel_capture_prompts``: that flag
    gates span attributes and is applied by the shipped ``SpanRecorder`` at
    the exporter, where the data actually leaves the process. Reusing it
    here would broaden its meaning from "do not put prompts in span
    attributes" to "do not give any consumer the text", silently emptying
    callback payloads that shipped code has always populated.

    ``FrameworkConfig.observability_redact_payloads`` is the explicit
    opt-in for adopters who want the bridge itself to withhold. It is read
    through ``getattr`` so a duck-typed config still works, but it is a
    declared field precisely because a ``getattr``-only control on an
    ``extra="ignore"`` model would be silently dropped and fail open.
    """
    if bool(getattr(config, "observability_redact_payloads", False)):
        return DENY_ALL
    return CAPTURE_ALL

output_text

output_text(text: str | None) -> str | None

Model-produced text, or None when withheld.

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

def output_text(self, text: str | None) -> str | None:
    """Model-produced text, or ``None`` when withheld."""
    return text if self.capture_outputs else None

prompt_text

prompt_text(text: str | None) -> str | None

The user-supplied prompt, or None when withheld.

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

def prompt_text(self, text: str | None) -> str | None:
    """The user-supplied prompt, or ``None`` when withheld."""
    return text if self.capture_prompts else None

tool_arguments

tool_arguments(arguments: Mapping[str, Any] | None) -> Mapping[str, Any] | None

Tool call arguments, or None when withheld.

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

def tool_arguments(
    self, arguments: Mapping[str, Any] | None
) -> Mapping[str, Any] | None:
    """Tool call arguments, or ``None`` when withheld."""
    return arguments if self.capture_tool_arguments else None