diff --git a/clearwing/__init__.py b/clearwing/__init__.py index 7e5132c8..0e0b5cd5 100644 --- a/clearwing/__init__.py +++ b/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() diff --git a/clearwing/agent/runtime.py b/clearwing/agent/runtime.py index 01e600f3..3bd19930 100644 --- a/clearwing/agent/runtime.py +++ b/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: diff --git a/clearwing/observability/integration.py b/clearwing/observability/integration.py index 4329a11f..ece74bde 100644 --- a/clearwing/observability/integration.py +++ b/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 diff --git a/clearwing/observability/phoenix.py b/clearwing/observability/phoenix.py index d43ac1d2..330c8a6a 100644 --- a/clearwing/observability/phoenix.py +++ b/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), diff --git a/clearwing/observability/telemetry.py b/clearwing/observability/telemetry.py index c8ba0fac..3d352089 100644 --- a/clearwing/observability/telemetry.py +++ b/clearwing/observability/telemetry.py @@ -112,6 +112,9 @@ 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. @@ -119,6 +122,12 @@ def record_llm_call( 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: diff --git a/clearwing/sourcehunt/hunter.py b/clearwing/sourcehunt/hunter.py index 02ab7e9f..0cc29e89 100644 --- a/clearwing/sourcehunt/hunter.py +++ b/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: diff --git a/clearwing/ui/web/app.py b/clearwing/ui/web/app.py index 990a0951..952f8d23 100644 --- a/clearwing/ui/web/app.py +++ b/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 # --------------------------------------------------------------- diff --git a/tests/test_observability.py b/tests/test_observability.py index 07b3bec4..2e837ce7 100644 --- a/tests/test_observability.py +++ b/tests/test_observability.py @@ -442,7 +442,7 @@ def test_phoenix_disabled_by_default(self): assert len(obs.tracer._exporters) == 1 def test_cost_update_emits_llm_span(self): - """cost_update event creates an llm_call span for Arize.""" + """cost_update event creates an llm_call span with OpenInference attrs.""" obs = ObservabilityIntegration() obs._on_cost_update({ "model": "claude-opus-4-6", @@ -457,10 +457,113 @@ def test_cost_update_emits_llm_span(self): llm_spans = obs._in_memory.get_spans("llm_call") assert len(llm_spans) == 1 s = llm_spans[0] - assert s.attributes["llm.model"] == "claude-opus-4-6" - assert s.attributes["llm.token_count.input"] == 2000 + # OpenInference semantic conventions: + # https://github.com/Arize-ai/openinference/blob/main/spec/semantic_conventions.md + assert s.attributes["openinference.span.kind"] == "LLM" + assert s.attributes["llm.model_name"] == "claude-opus-4-6" + assert s.attributes["llm.provider"] == "anthropic" + assert s.attributes["llm.token_count.prompt"] == 2000 + assert s.attributes["llm.token_count.completion"] == 400 + assert s.attributes["llm.token_count.total"] == 2400 + assert s.attributes["llm.token_count.cached"] == 500 assert s.attributes["llm.cost_usd"] == 0.12 + def test_cost_update_records_elapsed_and_provider(self): + """The emitted LLM span's duration reflects elapsed_ms.""" + obs = ObservabilityIntegration() + obs._on_cost_update({ + "model": "claude-sonnet-4-6", + "provider": "anthropic", + "input_tokens": 100, + "output_tokens": 50, + "total_cost_usd": 0.001, + "elapsed_ms": 1234, + }) + obs.tracer.flush() + llm_spans = obs._in_memory.get_spans("llm_call") + assert len(llm_spans) == 1 + s = llm_spans[0] + assert s.attributes["llm.provider"] == "anthropic" + # duration_ms should be within a small tolerance of elapsed_ms (span + # closed roughly at "now"; start backdated by elapsed_ms). + assert abs(s.duration_ms - 1234) < 500 + + def test_bootstrap_from_env_noop_without_endpoint(self, monkeypatch): + monkeypatch.delenv("PHOENIX_ENDPOINT", raising=False) + monkeypatch.delenv("PHOENIX_PROJECT", raising=False) + # Clear any singleton left over from a prior test. + if ObservabilityIntegration._singleton is not None: + ObservabilityIntegration._singleton.disconnect() + assert ObservabilityIntegration.bootstrap_from_env() is None + assert ObservabilityIntegration.bootstrap_from_env() is None + assert ObservabilityIntegration._singleton is None + + def test_bootstrap_from_env_connects_when_endpoint_set(self, monkeypatch): + monkeypatch.setenv("PHOENIX_ENDPOINT", "http://phoenix:6006") + monkeypatch.setenv("PHOENIX_PROJECT", "clearwing-test") + if ObservabilityIntegration._singleton is not None: + ObservabilityIntegration._singleton.disconnect() + first = ObservabilityIntegration.bootstrap_from_env() + try: + assert first is not None + assert first._connected is True + second = ObservabilityIntegration.bootstrap_from_env() + assert second is first + finally: + if first is not None: + first.disconnect() + + def test_bootstrap_from_env_disconnect_cleans_up(self, monkeypatch): + monkeypatch.setenv("PHOENIX_ENDPOINT", "http://phoenix:6006") + monkeypatch.setenv("PHOENIX_PROJECT", "clearwing-test") + if ObservabilityIntegration._singleton is not None: + ObservabilityIntegration._singleton.disconnect() + first = ObservabilityIntegration.bootstrap_from_env() + assert first is not None + first.disconnect() + assert ObservabilityIntegration._singleton is None + second = ObservabilityIntegration.bootstrap_from_env() + try: + assert second is not None + assert second is not first + finally: + if second is not None: + second.disconnect() + + def test_phoenix_span_kind_derived_from_openinference(self): + """PhoenixExporter's _to_readable_span picks OTel kind from attrs.""" + from opentelemetry.trace import SpanKind + + from clearwing.observability.phoenix import _span_kind_from_openinference + from clearwing.observability.tracer import Span + + llm_span = Span( + trace_id="t1", + span_id="s1", + name="llm_call", + attributes={"openinference.span.kind": "LLM"}, + ) + assert _span_kind_from_openinference(llm_span) == SpanKind.CLIENT + + tool_span = Span( + trace_id="t1", + span_id="s2", + name="tool", + attributes={"openinference.span.kind": "TOOL"}, + ) + assert _span_kind_from_openinference(tool_span) == SpanKind.CLIENT + + chain_span = Span( + trace_id="t1", + span_id="s3", + name="chain", + attributes={"openinference.span.kind": "CHAIN"}, + ) + assert _span_kind_from_openinference(chain_span) == SpanKind.INTERNAL + + unset_span = Span(trace_id="t1", span_id="s4", name="unset") + assert _span_kind_from_openinference(unset_span) == SpanKind.INTERNAL + # --------------------------------------------------------------------------- # PhoenixExporter tests diff --git a/tests/test_telemetry.py b/tests/test_telemetry.py index edc55d22..2dbe8073 100644 --- a/tests/test_telemetry.py +++ b/tests/test_telemetry.py @@ -1,5 +1,6 @@ """Tests for the CostTracker telemetry module.""" +from clearwing.core.events import EventBus, EventType from clearwing.observability.telemetry import CostSummary, CostTracker, ToolUsage @@ -131,6 +132,59 @@ def test_reset(self): assert t.tool_calls == 0 assert t.by_tool == {} + def test_record_llm_call_emits_cost_update_with_elapsed_and_provider(self): + """New keyword args ride along in the COST_UPDATE payload.""" + received: list[dict] = [] + + def handler(data): + received.append(data) + + bus = EventBus() + bus.subscribe(EventType.COST_UPDATE, handler) + try: + t = CostTracker() + t.record_llm_call( + 1000, + 500, + "claude-sonnet-4-6", + cached_tokens=100, + elapsed_ms=1234.5, + provider="anthropic", + ) + finally: + bus.unsubscribe(EventType.COST_UPDATE, handler) + + assert len(received) == 1 + payload = received[0] + assert payload["input_tokens"] == 1000 + assert payload["output_tokens"] == 500 + assert payload["cached_tokens"] == 100 + assert payload["model"] == "claude-sonnet-4-6" + assert payload["provider"] == "anthropic" + assert payload["elapsed_ms"] == 1234.5 + + def test_record_llm_call_backward_compatible_without_new_kwargs(self): + """Legacy callers without the new kwargs still emit a valid payload.""" + received: list[dict] = [] + + def handler(data): + received.append(data) + + bus = EventBus() + bus.subscribe(EventType.COST_UPDATE, handler) + try: + t = CostTracker() + t.record_llm_call(1000, 500, "claude-sonnet-4-6") + finally: + bus.unsubscribe(EventType.COST_UPDATE, handler) + + assert len(received) == 1 + payload = received[0] + # Falls back to safe defaults. + assert payload["provider"] == "unknown" + assert payload["elapsed_ms"] == 0 + assert payload["cached_tokens"] == 0 + def test_pricing_table(self): assert "claude-sonnet-4-6" in CostTracker.PRICING assert "claude-opus-4-6" in CostTracker.PRICING