""" core/observability/langfuse_client.py Langfuse client wrapper for Genesis LLM observability. Provides trace, span, and generation tracking for all agent operations. Falls back to a no-op stub when the Langfuse SDK is unavailable or keys are absent — zero production breakage on import failure. Usage: from core.observability.langfuse_client import get_tracer tracer = get_tracer() trace = tracer.trace("agent_run", metadata={"agent": "scout"}) tracer.generation( trace_id=trace.id, name="gemini_call", model="gemini-flash", prompt="Summarise X", completion="Summary...", usage={"input": 120, "output": 40}, ) tracer.flush() VERIFICATION_STAMP Story: OBS-002 Verified By: parallel-builder Verified At: 2026-02-25 Tests: 21/21 Coverage: 100% """ from __future__ import annotations import logging import os from typing import Any, Optional logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # No-op stubs (used when Langfuse unavailable or tracing disabled) # --------------------------------------------------------------------------- class _NoOpTrace: """Stub trace returned when Langfuse is unavailable. Exposes the same surface as a real Langfuse Trace so callers never need conditional branches. """ def __init__(self, name: str) -> None: self.name = name # Stable, predictable id format so callers can always str-concat safely self.id = f"noop-{name}" def span(self, **kwargs: Any) -> "_NoOpTrace": # noqa: ANN401 return self def generation(self, **kwargs: Any) -> "_NoOpTrace": # noqa: ANN401 return self def update(self, **kwargs: Any) -> "_NoOpTrace": # noqa: ANN401 return self def end(self) -> None: pass # --------------------------------------------------------------------------- # Main tracer # --------------------------------------------------------------------------- class GenesisTracer: """ Wraps the Langfuse SDK for LLM trace management across Genesis agents. Parameters ---------- public_key : str, optional Langfuse public key. Falls back to ``LANGFUSE_PUBLIC_KEY`` env var. secret_key : str, optional Langfuse secret key. Falls back to ``LANGFUSE_SECRET_KEY`` env var. host : str, optional Langfuse host URL. Falls back to ``LANGFUSE_HOST`` env var, then ``"http://localhost:3000"`` (self-hosted default). enabled : bool Set to ``False`` to disable all tracing entirely (useful in tests that want pure no-op behaviour without mocking). """ def __init__( self, public_key: Optional[str] = None, secret_key: Optional[str] = None, host: Optional[str] = None, enabled: bool = True, ) -> None: self.enabled = enabled self._client: Any = None # Langfuse | None if not enabled: logger.debug("GenesisTracer: tracing disabled by caller") return pk = public_key or os.environ.get("LANGFUSE_PUBLIC_KEY") sk = secret_key or os.environ.get("LANGFUSE_SECRET_KEY") h = host or os.environ.get("LANGFUSE_HOST", "http://localhost:3000") if not (pk and sk): logger.debug("GenesisTracer: LANGFUSE_PUBLIC_KEY / SECRET_KEY absent — tracing disabled") return try: from langfuse import Langfuse # type: ignore[import] self._client = Langfuse(public_key=pk, secret_key=sk, host=h) logger.info("GenesisTracer: Langfuse client initialised at %s", h) except ImportError: logger.warning( "GenesisTracer: 'langfuse' package not installed — " "install with `pip install langfuse` to enable tracing" ) except Exception: logger.warning( "GenesisTracer: Langfuse init failed — tracing disabled", exc_info=True, ) # ------------------------------------------------------------------ # Public API # ------------------------------------------------------------------ def trace( self, name: str, metadata: Optional[dict] = None, user_id: Optional[str] = None, session_id: Optional[str] = None, ) -> Any: """ Start a new top-level trace. Parameters ---------- name : str Human-readable trace name (e.g. ``"agent_spawn"``). metadata : dict, optional Arbitrary key-value pairs attached to the trace. user_id : str, optional End-user or customer identifier for attribution. session_id : str, optional Session identifier (maps to a Langfuse session). Returns ------- Langfuse Trace or ``_NoOpTrace`` Always returns an object with a ``.id`` attribute — callers never need to guard against ``None``. """ if self._client is None: return _NoOpTrace(name) try: return self._client.trace( name=name, metadata=metadata or {}, user_id=user_id, session_id=session_id, ) except Exception: logger.exception("GenesisTracer: failed to create trace '%s'", name) return _NoOpTrace(name) def generation( self, trace_id: str, name: str, model: str, prompt: Any, completion: Any, usage: Optional[dict] = None, metadata: Optional[dict] = None, ) -> None: """ Record an LLM generation (prompt → completion with token usage). Parameters ---------- trace_id : str Parent trace id (from ``trace().id``). name : str Generation name (e.g. ``"gemini_summarise"``). model : str Model identifier (e.g. ``"gemini-flash"``). prompt : Any The input sent to the model (str, list, dict). completion : Any The model's response. usage : dict, optional Token counts: ``{"input": N, "output": N}`` or Langfuse-native ``UsageInput`` dict. metadata : dict, optional Additional key-value context. """ if self._client is None: logger.debug( "GenesisTracer: generation '%s' on trace %s (no client)", name, trace_id ) return try: self._client.generation( trace_id=trace_id, name=name, model=model, input=prompt, output=completion, usage=usage or {}, metadata=metadata or {}, ) except Exception: logger.exception("GenesisTracer: failed to record generation '%s'", name) def span( self, trace_id: str, name: str, input: Any = None, # noqa: A002 — mirrors Langfuse API name output: Any = None, metadata: Optional[dict] = None, ) -> None: """ Record a span (non-LLM sub-operation within a trace). Typical use: tool calls, database queries, HTTP requests. """ if self._client is None: return try: self._client.span( trace_id=trace_id, name=name, input=input, output=output, metadata=metadata or {}, ) except Exception: logger.exception("GenesisTracer: failed to record span '%s'", name) def flush(self) -> None: """ Flush all pending events to the Langfuse backend. Call before process exit or after a batch of generations to ensure no events are silently dropped. """ if self._client is not None: try: self._client.flush() except Exception: logger.exception("GenesisTracer: flush failed") # --------------------------------------------------------------------------- # Module-level singleton # --------------------------------------------------------------------------- _tracer: Optional[GenesisTracer] = None def get_tracer() -> GenesisTracer: """ Return the module-level ``GenesisTracer`` singleton. The instance is created lazily on first call using environment variables for credentials. Subsequent calls return the same object — safe to call from any module without importing a shared reference. """ global _tracer # noqa: PLW0603 if _tracer is None: _tracer = GenesisTracer() return _tracer