observability: auto-wire Phoenix from env + fix OpenInference span rendering

clearwing/__init__.py

@@ -9,7 +9,13 @@
 
 def main():
     """Main entry point for Clearwing."""
+    from .observability.integration import ObservabilityIntegration
     from .ui.cli import CLI
 
+    # Auto-wire Phoenix / OTLP tracing when PHOENIX_ENDPOINT is set. No-op
+    # otherwise, so plain `clearwing …` invocations without the env vars pay
+    # zero cost.
+    ObservabilityIntegration.bootstrap_from_env()
+
     cli = CLI()
     cli.run()

clearwing/agent/runtime.py

@@ -258,12 +258,16 @@ async def _aassistant_step(self, state: dict[str, Any]) -> dict[str, Any]:
         system, chat_messages = _coerce_chat_messages(messages)
         system = "\n\n".join(part for part in (sys_prompt, system) if part) or sys_prompt
 
+        import time as _time
+
+        _llm_start = _time.perf_counter()
         response = await self.llm.achat_stream(
             messages=chat_messages,
             system=system,
             tools=self.native_tools or None,
             on_text_delta=self.on_text_delta,
         )
+        _llm_elapsed_ms = (_time.perf_counter() - _llm_start) * 1000.0
 
         assistant_text = response_text(response)
         # tool_calls are raw genai ToolCall objects (.call_id/.fn_name/
@@ -289,7 +293,13 @@ async def _aassistant_step(self, state: dict[str, Any]) -> dict[str, Any]:
         input_tokens = (usage.prompt_tokens or 0) if usage else 0
         output_tokens = (usage.completion_tokens or 0) if usage else 0
         if self.cost_tracker and (input_tokens or output_tokens):
-            self.cost_tracker.record_llm_call(input_tokens, output_tokens, self.model_name)
+            self.cost_tracker.record_llm_call(
+                input_tokens,
+                output_tokens,
+                self.model_name,
+                elapsed_ms=_llm_elapsed_ms,
+                provider=getattr(self.llm, "provider_name", None),
+            )
             state["total_cost_usd"] = self.cost_tracker.total_cost_usd
             state["total_tokens"] = self.cost_tracker.input_tokens + self.cost_tracker.output_tokens
             if self.audit_logger:

clearwing/observability/integration.py

@@ -3,7 +3,8 @@
 from __future__ import annotations
 
 import logging
-from typing import Any
+import os
+from typing import Any, ClassVar
 
 from clearwing.core.events import EventBus, EventType
 
@@ -32,8 +33,16 @@ class ObservabilityIntegration:
         # ... run agent ...
         obs.disconnect()
         print(obs.metrics.format_prometheus())
+
+    For env-driven auto-wiring at process startup, prefer
+    :meth:`bootstrap_from_env` — it is a no-op unless Phoenix env vars are set.
     """
 
+    # Process-wide singleton established by :meth:`bootstrap_from_env` so
+    # startup hooks in the CLI and webui don't double-subscribe. Tests may
+    # clear this by calling ``disconnect()``, which unsets the reference.
+    _singleton: ClassVar["ObservabilityIntegration | None"] = None
+
     def __init__(self, debug: bool = False, exporters: list = None):
         if exporters is None:
             exporters = []
@@ -49,6 +58,26 @@ def __init__(self, debug: bool = False, exporters: list = None):
         self._connected = False
         self._handlers = {}
 
+    @classmethod
+    def bootstrap_from_env(cls) -> "ObservabilityIntegration | None":
+        """Idempotently instantiate + connect when Phoenix env vars are set.
+
+        Returns the shared singleton, or ``None`` if
+        ``PHOENIX_ENDPOINT`` / ``PHOENIX_PROJECT`` are unset. Safe to call
+        from every process entry point (CLI ``main()``, FastAPI
+        ``create_app()``, machine-fd subcommand); subsequent calls return the
+        already-connected instance without re-subscribing to the EventBus.
+        """
+        if cls._singleton is not None:
+            return cls._singleton
+        if not os.environ.get("PHOENIX_ENDPOINT") or not os.environ.get("PHOENIX_PROJECT"):
+            return None
+        instance = cls()
+        instance.connect()
+        cls._singleton = instance
+        logger.info("ObservabilityIntegration bootstrapped from env")
+        return instance
+
     def connect(self) -> None:
         """Subscribe to EventBus events."""
         if self._connected:
@@ -80,6 +109,10 @@ def disconnect(self) -> None:
 
         self.tracer.shutdown()
         self._connected = False
+        # Release the bootstrap singleton so a follow-up bootstrap can create
+        # a fresh instance (primarily useful for tests).
+        if type(self)._singleton is self:
+            type(self)._singleton = None
 
     @property
     def spans(self) -> list:
@@ -113,26 +146,37 @@ def _on_cost_update(self, data: Any) -> None:
         if not isinstance(data, dict):
             return
         model = data.get("model", "unknown")
+        input_tokens = int(data.get("input_tokens", 0) or 0)
+        output_tokens = int(data.get("output_tokens", 0) or 0)
+        cached_tokens = int(data.get("cached_tokens", 0) or 0)
+
         self.metrics.increment("llm_calls_total", labels={"model": model})
-        self.metrics.increment("input_tokens_total", value=float(data.get("input_tokens", 0)))
-        self.metrics.increment("output_tokens_total", value=float(data.get("output_tokens", 0)))
+        self.metrics.increment("input_tokens_total", value=float(input_tokens))
+        self.metrics.increment("output_tokens_total", value=float(output_tokens))
         self.metrics.set_gauge("total_cost_usd", data.get("total_cost_usd", 0.0))
 
-        # Emit a synthetic span so Arize sees per-call LLM telemetry
+        # Emit a synthetic span so Arize Phoenix sees per-call LLM telemetry.
+        # Attribute names follow the OpenInference semantic conventions —
+        # https://github.com/Arize-ai/openinference/blob/main/spec/semantic_conventions.md
+        # — so Phoenix renders the span in its LLM view rather than as a
+        # generic "internal" span.
         import time
 
         now = time.time()
-        elapsed_s = data.get("elapsed_ms", 0) / 1000.0
-        with self.tracer.span("llm_call", attributes={
-            "llm.model": model,
+        elapsed_ms = data.get("elapsed_ms", 0) or 0
+        elapsed_s = elapsed_ms / 1000.0
+        attributes = {
+            "openinference.span.kind": "LLM",
+            "llm.model_name": model,
             "llm.provider": data.get("provider", "unknown"),
-            "llm.token_count.input": data.get("input_tokens", 0),
-            "llm.token_count.output": data.get("output_tokens", 0),
-            "llm.token_count.cached": data.get("cached_tokens", 0),
+            "llm.token_count.prompt": input_tokens,
+            "llm.token_count.completion": output_tokens,
+            "llm.token_count.total": input_tokens + output_tokens,
+            "llm.token_count.cached": cached_tokens,
             "llm.cost_usd": data.get("total_cost_usd", 0.0),
-            "span.kind": "llm",
-        }) as s:
-            # Backdate start to reflect actual call timing
+        }
+        with self.tracer.span("llm_call", attributes=attributes) as s:
+            # Backdate start so the span duration reflects actual call latency.
             if elapsed_s > 0:
                 s.start_time = now - elapsed_s
 

clearwing/observability/phoenix.py

@@ -15,10 +15,29 @@
 logger = logging.getLogger(__name__)
 
 
+def _span_kind_from_openinference(span: Span):
+    """Derive OTel SpanKind from an ``openinference.span.kind`` attribute.
+
+    Phoenix's UI keys off both the attribute string and the OTel ``SpanKind``
+    enum. We keep the attribute on the outgoing span (so Phoenix classifies it
+    correctly) and additionally translate to the most fitting OTel kind:
+
+    - ``"LLM"`` / ``"TOOL"``  → ``SpanKind.CLIENT`` (outbound call)
+    - ``"CHAIN"`` / ``"AGENT"`` → ``SpanKind.INTERNAL`` (internal orchestration)
+    - unset / anything else   → ``SpanKind.INTERNAL``
+    """
+    from opentelemetry.trace import SpanKind
+
+    kind = span.attributes.get("openinference.span.kind")
+    if kind in ("LLM", "TOOL"):
+        return SpanKind.CLIENT
+    return SpanKind.INTERNAL
+
+
 def _to_readable_span(span: Span, resource):
     """Bridge a Clearwing Span to an OTel ReadableSpan."""
     from opentelemetry.sdk.trace import ReadableSpan as _ReadableSpan
-    from opentelemetry.trace import SpanContext, SpanKind, TraceFlags
+    from opentelemetry.trace import SpanContext, TraceFlags
     from opentelemetry.trace.status import Status, StatusCode
 
     trace_id = int(span.trace_id, 16) & ((1 << 128) - 1)
@@ -37,7 +56,7 @@ def _to_readable_span(span: Span, resource):
         resource=resource,
         attributes=span.attributes,
         events=tuple(),
-        kind=SpanKind.INTERNAL,
+        kind=_span_kind_from_openinference(span),
         status=status,
         start_time=int(span.start_time * 1e9),
         end_time=int(span.end_time * 1e9),

clearwing/observability/telemetry.py

@@ -112,13 +112,22 @@ def record_llm_call(
         output_tokens: int,
         model: str,
         cached_tokens: int = 0,
+        *,
+        elapsed_ms: float | None = None,
+        provider: str | None = None,
     ) -> None:
         """Record token usage for a single LLM call and update the running cost.
 
         If *model* is not present in the pricing table the default Sonnet
         pricing is used.  ``cached_tokens`` bills at the model's cached rate.
         When an ``EventBus`` is available a ``COST_UPDATE`` event is emitted
         after updating counters.
+
+        ``elapsed_ms`` (wall-clock latency of the call) and ``provider`` are
+        optional; when supplied they ride along in the ``COST_UPDATE`` payload
+        so ``ObservabilityIntegration`` can attach them to the synthetic
+        Phoenix LLM span. Keyword-only to keep call sites explicit and future
+        additions non-breaking.
         """
         cost = self.estimate_cost(input_tokens, output_tokens, model, cached_tokens)
 
@@ -128,14 +137,17 @@ def record_llm_call(
             self.total_cost_usd += cost
 
         try:
-            EventBus.emit(
+            EventBus().emit(
                 EventType.COST_UPDATE,
                 {
                     "input_tokens": input_tokens,
                     "output_tokens": output_tokens,
+                    "cached_tokens": cached_tokens,
                     "cost": cost,
                     "total_cost_usd": self.total_cost_usd,
                     "model": model,
+                    "provider": provider or "unknown",
+                    "elapsed_ms": elapsed_ms or 0,
                 },
             )
         except Exception:

clearwing/sourcehunt/hunter.py

@@ -1456,11 +1456,15 @@ async def arun(self) -> HunterRunResult:
                     messages = await self.summarizer.summarize(messages, self.llm)
                     logger.info("Hunter context summarized: %d → %d messages", pre, len(messages))
 
+                import time as _time
+
+                _llm_start = _time.perf_counter()
                 response = await self.llm.achat(
                     messages=messages,
                     system=self.prompt,
                     tools=self.tools,
                 )
+                _llm_elapsed_ms = (_time.perf_counter() - _llm_start) * 1000.0
             # Preserve the provider's reasoning_content alongside the
             # visible text. `response.first_text` only returns the
             # first Text part — reasoning/thinking blocks are separate
@@ -1495,12 +1499,25 @@ async def arun(self) -> HunterRunResult:
             # response where nothing was cache-served.
             details = getattr(response.usage, "prompt_tokens_details", None)
             cached_tokens = (getattr(details, "cached_tokens", None) or 0) if details else 0
+            _prompt_toks = response.usage.prompt_tokens or 0
+            _completion_toks = response.usage.completion_tokens or 0
             total_cost_usd += _estimate_cost_usd(
-                response.usage.prompt_tokens or 0,
-                response.usage.completion_tokens or 0,
+                _prompt_toks,
+                _completion_toks,
                 self.llm.model_name,
                 cached_tokens,
             )
+            # Also push through the singleton so ObservabilityIntegration
+            # emits a Phoenix LLM span for this hunter call.
+            if _prompt_toks or _completion_toks:
+                CostTracker().record_llm_call(
+                    _prompt_toks,
+                    _completion_toks,
+                    self.llm.model_name,
+                    cached_tokens=cached_tokens,
+                    elapsed_ms=_llm_elapsed_ms,
+                    provider=getattr(self.llm, "provider_name", None),
+                )
 
             last_assistant_text = response.first_text or ""
             if last_assistant_text:

clearwing/ui/web/app.py

@@ -21,6 +21,7 @@
 from clearwing.agent.runtime import Command
 from clearwing.core.events import EventBus, EventType
 from clearwing.observability import MetricsCollector
+from clearwing.observability.integration import ObservabilityIntegration
 
 logger = logging.getLogger(__name__)
 
@@ -54,6 +55,14 @@ def create_app():
         allow_headers=["*"],
     )
 
+    # Auto-wire Phoenix / OTLP tracing when PHOENIX_ENDPOINT is set. Flush the
+    # OTel batch processor on graceful shutdown so nothing gets dropped.
+    _obs = ObservabilityIntegration.bootstrap_from_env()
+    if _obs is not None:
+        @app.on_event("shutdown")
+        def _flush_observability() -> None:
+            _obs.disconnect()
+
     # ---------------------------------------------------------------
     # Authentication for privileged endpoints
     # ---------------------------------------------------------------