diff --git a/src/llm/executor.py b/src/llm/executor.py index a9fe3703..017ce669 100644 --- a/src/llm/executor.py +++ b/src/llm/executor.py @@ -19,7 +19,7 @@ from typing import Any, Literal, TypeVar, overload from pydantic import BaseModel -from src.config import ModelConfig, ModelTransport +from src.config import ModelConfig, ModelTransport, settings from src.telemetry.logging import conditional_observe from .backend import CompletionResult as BackendCompletionResult @@ -29,6 +29,7 @@ from .registry import CLIENTS, backend_for_provider from .request_builder import execute_completion, execute_stream from .runtime import ( AttemptPlan, + annotate_current_generation_io, annotate_current_langfuse_trace, effective_config_for_call, ) @@ -45,6 +46,82 @@ logger = logging.getLogger(__name__) M = TypeVar("M", bound=BaseModel) +# ModelConfig fields that must NEVER reach a trace: secrets and nested holders +# of secrets. Everything else on the config is a safe tuning knob and is dumped +# automatically — so new knobs get traced without touching this code. Keep this +# a deny-list (small, stable) rather than an allow-list (drifts with the model). +_UNSAFE_CONFIG_FIELDS = frozenset( + { + "api_key", # provider secret + "base_url", # may embed credentials / private host + "fallback", # ResolvedFallbackConfig carries its own api_key/base_url + "provider_params", # opaque dict; can carry auth headers/keys + } +) + + +def _langfuse_model_parameters( + *, + max_tokens: int, + config: ModelConfig, + json_mode: bool, + verbosity: str | None, + stream: bool, + tools: list[dict[str, Any]] | None, + tool_choice: str | dict[str, Any] | None, + response_model: type[BaseModel] | None, +) -> dict[str, Any]: + """Serializable tuning knobs for the Langfuse generation. + + Surfaces everything @observe auto-capture used to show (temperature, tools, + ...) MINUS the live client and the secret-bearing config fields. We dump the + resolved `effective_config` and deny-list only `_UNSAFE_CONFIG_FIELDS`, so a + new ModelConfig knob is traced automatically — no allow-list to keep in sync. + `mode="json"` coerces enums/sub-models to JSON-safe values. See HONCHO-4HA. + """ + params: dict[str, Any] = config.model_dump( + exclude=set(_UNSAFE_CONFIG_FIELDS), exclude_none=True, mode="json" + ) + # Per-call extras that live outside ModelConfig. + params["max_tokens"] = max_tokens + params["stream"] = stream + params["json_mode"] = json_mode + if verbosity is not None: + params["verbosity"] = verbosity + if response_model is not None: + params["response_format"] = response_model.__name__ + if tools: + params["tools"] = [ + t.get("name") or t.get("function", {}).get("name") or "unknown" + for t in tools + ] + if tool_choice is not None: + params["tool_choice"] = ( + tool_choice if isinstance(tool_choice, str) else str(tool_choice) + ) + return params + + +def _langfuse_usage_details(response: HonchoLLMCallResponse[Any]) -> dict[str, int]: + """Token usage duplicated onto the Langfuse generation. + + These counts are also emitted via CloudEvents (LLMCallCompletedEvent), but + we mirror them here so Langfuse renders per-call tokens + cost natively, + including Anthropic-style prompt-cache reads/writes. Zero-valued cache keys + are dropped so non-cached calls stay tidy. Stream calls don't surface token + totals at this layer, so usage is set only on the non-stream path. + """ + usage: dict[str, int] = { + "input": response.input_tokens, + "output": response.output_tokens, + } + if response.cache_read_input_tokens: + usage["cache_read_input_tokens"] = response.cache_read_input_tokens + if response.cache_creation_input_tokens: + usage["cache_creation_input_tokens"] = response.cache_creation_input_tokens + return usage + + def _outcome_from_error( err: BaseException | None, ) -> Literal["success", "error", "cancelled"]: @@ -266,7 +343,18 @@ async def honcho_llm_call_inner( ) -> AsyncIterator[HonchoLLMCallStreamChunk]: ... -@conditional_observe(name="LLM Call", as_type="generation") +@conditional_observe( + name="LLM Call", + as_type="generation", + # Disable @observe auto-capture: it would serialize `client_override` (a + # live AsyncOpenAI/genai client) and `selected_config` (carries api_key) + # into the span. Auto-capture deep-copies the client into a half-built + # object whose teardown raises `_state`/`_http_options` AttributeErrors + # (HONCHO-4HA) and leaks the key. We set curated input/output explicitly + # below via `annotate_current_generation_io`, preserving full fidelity. + capture_input=False, + capture_output=False, +) async def honcho_llm_call_inner( provider: ModelTransport, model: str, @@ -331,6 +419,26 @@ async def honcho_llm_call_inner( thinking_budget_tokens=thinking_budget_tokens, reasoning_effort=reasoning_effort, ) + + # Explicit generation input + tuning knobs (replaces @observe auto-capture, + # which would serialize the live client / api key). Set before the stream + # branch so it lands on the generation span for both paths. Guard on the + # public key so we don't build the (model_dump-backed) payload when Langfuse + # is disabled — the annotate helper no-ops, but the payload still costs. + if settings.LANGFUSE_PUBLIC_KEY: + annotate_current_generation_io( + input=messages, + model_parameters=_langfuse_model_parameters( + max_tokens=max_tokens, + config=effective_config, + json_mode=json_mode, + verbosity=verbosity, + stream=stream, + tools=tools, + tool_choice=tool_choice, + response_model=response_model, + ), + ) # json_mode + verbosity are per-call transport toggles, not ModelConfig # knobs — they pass through extra_params. execute_completion merges # build_config_extra_params(effective_config) on top for top_p/seed/etc. @@ -416,7 +524,17 @@ async def honcho_llm_call_inner( cache_policy=effective_config.cache_policy, extra_params=call_extras, ) - return completion_result_to_response(backend_result) + response = completion_result_to_response(backend_result) + # Explicit generation output + token usage (replaces @observe + # auto-capture). The stream path closes this span before drain, so its + # output is stamped on the run-level span instead + # (StreamingResponseWithMetadata). + if settings.LANGFUSE_PUBLIC_KEY: + annotate_current_generation_io( + output=response, + usage_details=_langfuse_usage_details(response), + ) + return response except BaseException as exc: error = exc raise diff --git a/src/llm/runtime.py b/src/llm/runtime.py index ed01229d..ec07acf8 100644 --- a/src/llm/runtime.py +++ b/src/llm/runtime.py @@ -102,6 +102,48 @@ def annotate_current_langfuse_trace( logger.debug("Failed to update Langfuse trace metadata: %s", exc) +def annotate_current_generation_io( + *, + input: Any = None, # noqa: A002 - mirrors langfuse's `input` kwarg name + output: Any = None, + model_parameters: dict[str, Any] | None = None, + usage_details: dict[str, Any] | None = None, +) -> None: + """Set explicit input/output/model_parameters/usage on the current generation. + + Used in place of ``@observe``'s auto-capture (disabled on + ``honcho_llm_call_inner``) so the provider client and api-key-bearing + ``ModelConfig`` arguments are never serialized into traces. Auto-capture + deep-copies those args, producing half-constructed clients whose teardown + raised ``AsyncHttpxClientWrapper ... no attribute '_state'`` / + ``BaseApiClient ... no attribute '_http_options'`` (HONCHO-4HA) and leaked + ``ModelConfig.api_key``. We instead hand Langfuse curated, serializable + values: ``messages`` in, response out, and the call's tuning knobs as + ``model_parameters`` — preserving (and tidying) full trace fidelity. + + Best-effort: telemetry must never fail the LLM call. + """ + if not settings.LANGFUSE_PUBLIC_KEY: + return + payload: dict[str, Any] = {} + if input is not None: + payload["input"] = input + if output is not None: + payload["output"] = output + if model_parameters: + payload["model_parameters"] = model_parameters + if usage_details: + payload["usage_details"] = usage_details + if not payload: + return + try: + from langfuse import get_client + + get_client().update_current_generation(**payload) + except Exception as exc: # pragma: no cover - best-effort telemetry + logger.debug("Failed to set Langfuse generation IO: %s", exc) + + def _base_metadata(telemetry: LLMTelemetryContext) -> dict[str, str]: """Static routing/attribution metadata (everything except ``iteration``). @@ -480,6 +522,7 @@ __all__ = [ "AttemptPlan", "LangfuseAgentRun", "LangfuseAgentStep", + "annotate_current_generation_io", "annotate_current_langfuse_trace", "current_attempt", "effective_config_for_call", diff --git a/src/telemetry/logging.py b/src/telemetry/logging.py index 441b1f96..b653952e 100644 --- a/src/telemetry/logging.py +++ b/src/telemetry/logging.py @@ -61,6 +61,8 @@ def conditional_observe( *, name: str | None = None, as_type: ObserveAsType | None = None, + capture_input: bool | None = None, + capture_output: bool | None = None, ) -> Callable[[Callable[P, R]], Callable[P, R]]: ... @@ -69,6 +71,8 @@ def conditional_observe( *, name: str | None = None, as_type: ObserveAsType | None = None, + capture_input: bool | None = None, + capture_output: bool | None = None, ) -> Callable[P, R] | Callable[[Callable[P, R]], Callable[P, R]]: """ Conditionally apply the @observe decorator only when LANGFUSE_PUBLIC_KEY is present. @@ -82,19 +86,35 @@ def conditional_observe( name: Optional name for the observation (when used as @conditional_observe(name="...")) as_type: Optional Langfuse observation type (e.g. "generation", "tool"). When omitted, Langfuse infers a default span. + capture_input: When ``False``, Langfuse does NOT auto-serialize the + function's arguments into the span input. Set this on functions that + receive live SDK clients or secret-bearing config as parameters + (e.g. the LLM executor): auto-capture would deep-copy those clients + into throwaway, half-constructed objects whose GC raises + ``AsyncHttpxClientWrapper ... no attribute '_state'`` / + ``BaseApiClient ... no attribute '_http_options'`` (see HONCHO-4HA), + and would also leak ``ModelConfig.api_key`` into traces. Pair with an + explicit ``update_current_generation(input=...)`` call to keep + full-fidelity input. ``None`` leaves the SDK default (capture on). + capture_output: When ``False``, Langfuse does NOT auto-serialize the + return value. Pair with an explicit + ``update_current_generation(output=...)``. ``None`` = SDK default. Returns: The decorated function if Langfuse is configured, otherwise the original function """ def decorator(f: Callable[P, R]) -> Callable[P, R]: - if settings.LANGFUSE_PUBLIC_KEY: - observe_name = name if name is not None else f.__name__ - if as_type is not None: - return observe(name=observe_name, as_type=as_type)(f) - return observe(name=observe_name)(f) - else: + if not settings.LANGFUSE_PUBLIC_KEY: return f + # `observe` treats None as "use SDK default", so passing the optionals + # straight through is equivalent to omitting them. + return observe( + name=name if name is not None else f.__name__, + as_type=as_type, + capture_input=capture_input, + capture_output=capture_output, + )(f) if func is not None: # Used as @conditional_observe (without parentheses) diff --git a/tests/utils/test_clients.py b/tests/utils/test_clients.py index 1567eef8..319e5f1f 100644 --- a/tests/utils/test_clients.py +++ b/tests/utils/test_clients.py @@ -960,11 +960,29 @@ class TestMainLLMCallFunction: assert captured["metadata"]["provider"] == "anthropic" assert captured["metadata"]["model"] == "claude-4-sonnet" # ...and the generation is named + carries per-call model/metadata. - mock_langfuse_client.update_current_generation.assert_called_once() - gen_kwargs = mock_langfuse_client.update_current_generation.call_args.kwargs - assert gen_kwargs["name"] == "Dialectic Agent LLM call" - assert gen_kwargs["model"] == "claude-4-sonnet" - assert gen_kwargs["metadata"]["provider"] == "anthropic" + gen_calls = mock_langfuse_client.update_current_generation.call_args_list + meta_kwargs = next(c.kwargs for c in gen_calls if "model" in c.kwargs) + assert meta_kwargs["name"] == "Dialectic Agent LLM call" + assert meta_kwargs["model"] == "claude-4-sonnet" + assert meta_kwargs["metadata"]["provider"] == "anthropic" + # Input/output are stamped explicitly: @observe auto-capture is + # disabled so the live client / api-key-bearing config never reach + # the trace (HONCHO-4HA), with no loss of trace fidelity. + input_kwargs = next(c.kwargs for c in gen_calls if "input" in c.kwargs) + assert input_kwargs["input"] == [{"role": "user", "content": "Hello"}] + output_kwargs = next(c.kwargs for c in gen_calls if "output" in c.kwargs) + assert output_kwargs["output"].content == "Named response" + # Token usage is duplicated onto the generation (also in CloudEvents) + # so Langfuse renders native per-call tokens + cost. + assert output_kwargs["usage_details"]["input"] == 5 + assert output_kwargs["usage_details"]["output"] == 5 + # Tuning knobs are tracked as model_parameters (not the live client + # or api-key-bearing config). No serialized client/secret anywhere. + params = next(c.kwargs for c in gen_calls if "model_parameters" in c.kwargs) + assert params["model_parameters"]["max_tokens"] == 100 + assert params["model_parameters"]["stream"] is False + assert "client_override" not in params["model_parameters"] + assert "api_key" not in params["model_parameters"] async def test_no_telemetry_still_stamps_trace_without_name(self): """Without telemetry, propagate_attributes still fires with namespace @@ -1010,10 +1028,68 @@ class TestMainLLMCallFunction: assert captured["metadata"]["provider"] == "anthropic" # Generation gets model + metadata even without a track_name — only # the name kwarg stays None. - mock_langfuse_client.update_current_generation.assert_called_once() - gen_kwargs = mock_langfuse_client.update_current_generation.call_args.kwargs - assert gen_kwargs["name"] is None - assert gen_kwargs["model"] == "claude-4-sonnet" + gen_calls = mock_langfuse_client.update_current_generation.call_args_list + meta_kwargs = next(c.kwargs for c in gen_calls if "model" in c.kwargs) + assert meta_kwargs["name"] is None + assert meta_kwargs["model"] == "claude-4-sonnet" + # Input/output stamped explicitly (auto-capture disabled; HONCHO-4HA). + input_kwargs = next(c.kwargs for c in gen_calls if "input" in c.kwargs) + assert input_kwargs["input"] == [{"role": "user", "content": "Hello"}] + output_kwargs = next(c.kwargs for c in gen_calls if "output" in c.kwargs) + assert output_kwargs["output"].content == "Unnamed response" + + +class TestLangfuseModelParameters: + """`_langfuse_model_parameters` is the deny-list seam that keeps secrets and + live clients out of Langfuse traces while still surfacing every tuning knob + (HONCHO-4HA). It dumps the config and excludes only secret-bearing fields, so + new knobs are traced automatically without an allow-list to maintain.""" + + def test_secret_fields_never_leak_but_knobs_do(self): + from src.config import ModelConfig + from src.llm.executor import ( + _langfuse_model_parameters, # pyright: ignore[reportPrivateUsage] + ) + + # A config shaped like the production override path: real api_key / + # base_url / nested fallback / opaque provider_params. + config = ModelConfig( + model="gpt-4o", + transport="openai", + api_key="sk-super-secret", + base_url="https://user:pw@private.host/v1", + temperature=0.7, + provider_params={"x-internal-auth": "leak-me"}, + ) + + params = _langfuse_model_parameters( + max_tokens=256, + config=config, + json_mode=True, + verbosity=None, + stream=False, + tools=[{"name": "search_memory"}], + tool_choice="auto", + response_model=None, + ) + + # Secrets and their nested holders are excluded entirely... + assert "api_key" not in params + assert "base_url" not in params + assert "fallback" not in params + assert "provider_params" not in params + # ...and no value anywhere echoes a secret. + flat = str(params) + assert "sk-super-secret" not in flat + assert "leak-me" not in flat + assert "private.host" not in flat + # Tuning knobs (config-derived + per-call) are still tracked. + assert params["model"] == "gpt-4o" + assert params["temperature"] == 0.7 + assert params["max_tokens"] == 256 + assert params["json_mode"] is True + assert params["tools"] == ["search_memory"] + assert params["tool_choice"] == "auto" class TestEdgeCases: